ship(P01): iter.Seq streaming merged into v0.3 milestone

---ci---
project: orca
phase: 1
milestone: v0.3
status: complete
requirements:
  covered: [REQ-022, REQ-030]
  partial: []
---/ci---

P01: iter.Seq streaming for --watch flags.
- JobRepo.Watch / NodeRepo.Watch: pull-based iter.Seq[[]*T] snapshot-per-tick (G-001)
- Immediate first yield before ticker (G-002)
- Table mode: clear-screen + re-render on change
- JSON mode: init/update/delete events, one line per change
- signal.NotifyContext on SIGINT/SIGTERM (D-023)
- 16 tests (8 store + 8 CLI), all pass under -race
- 4-layer verification passed

.ciagent/CHECKPOINT.json

@@ -1,9 +1,9 @@
 {
-  "phase": 0,
-  "stage": "grill",
+  "phase": 1,
+  "stage": "verify",
   "milestone": "v0.3",
   "milestone_slug": "scheduling-streaming",
-  "phase_role": "pre_execution",
+  "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-08-01T00:04:00Z"
+  "updated_at": "2026-08-01T00:10:00Z"
 }

internal/cli/job.go

@@ -5,6 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"github.com/google/uuid"
@@ -36,6 +39,7 @@ var (
 	stopID    string
 	runTarget string
 	runIDKey  string
+	jobWatch  bool
 )
 
 var jobRunCmd = &cobra.Command{
@@ -112,8 +116,11 @@ var jobRunCmd = &cobra.Command{
 var jobListCmd = &cobra.Command{
 	Use:   "list",
 	Short: "List all jobs",
-	Long:  "Display all jobs and their status.",
+	Long:  "Display all jobs and their status. Use --watch to stream updates until Ctrl-C.",
 	RunE: func(cmd *cobra.Command, args []string) error {
+		if jobWatch {
+			return watchJobs(cmd)
+		}
 		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
 		defer cancel()
 
@@ -142,6 +149,73 @@ var jobListCmd = &cobra.Command{
 	},
 }
 
+func watchJobs(cmd *cobra.Command) error {
+	ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+	return watchJobsCtx(cmd, ctx)
+}
+
+func watchJobsCtx(cmd *cobra.Command, ctx context.Context) error {
+	db, closer, err := openDB()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	out := cmd.OutOrStdout()
+
+	if jsonOutput {
+		seen := make(map[string]string)
+		for snapshot := range store.NewJobRepo(db).Watch(ctx) {
+			current := make(map[string]bool, len(snapshot))
+			for _, j := range snapshot {
+				current[j.ID] = true
+				compact, _ := json.Marshal(j)
+				key := string(compact)
+				if prev, ok := seen[j.ID]; !ok || prev != key {
+					event := "init"
+					if ok {
+						event = "update"
+					}
+					line, _ := json.Marshal(map[string]any{"event": event, "job": j})
+					fmt.Fprintln(out, string(line))
+					seen[j.ID] = key
+				}
+			}
+			for id := range seen {
+				if !current[id] {
+					line, _ := json.Marshal(map[string]any{"event": "delete", "id": id})
+					fmt.Fprintln(out, string(line))
+					delete(seen, id)
+				}
+			}
+		}
+		return nil
+	}
+
+	prevTable := ""
+	for snapshot := range store.NewJobRepo(db).Watch(ctx) {
+		table := renderJobTable(snapshot)
+		if table != prevTable {
+			fmt.Fprint(out, "\033[2J\033[H")
+			fmt.Fprint(out, table)
+			prevTable = table
+		}
+	}
+	return nil
+}
+
+func renderJobTable(jobs []*model.Job) string {
+	if len(jobs) == 0 {
+		return "No jobs.\n"
+	}
+	out := fmt.Sprintf("%-36s  %-20s  %-12s  %-8s\n", "ID", "NAME", "STATUS", "EXIT")
+	for _, j := range jobs {
+		out += fmt.Sprintf("%-36s  %-20s  %-12s  %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
+	}
+	return out
+}
+
 var jobStopCmd = &cobra.Command{
 	Use:   "stop [job-id]",
 	Short: "Stop a running job",
@@ -235,6 +309,7 @@ func init() {
 	jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id")
 	jobRunCmd.Flags().StringVar(&runTarget, "target", "", "pin job to a specific node id (overrides bin-packing)")
 	jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe")
+	jobListCmd.Flags().BoolVar(&jobWatch, "watch", false, "stream jobs until Ctrl-C (table refresh or --json per-event)")
 
 	jobCmd.AddCommand(jobRunCmd)
 	jobCmd.AddCommand(jobListCmd)

internal/cli/node.go

@@ -3,10 +3,13 @@ package cli
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"log/slog"
 	"os"
+	"os/signal"
 	"path/filepath"
+	"syscall"
 	"time"
 
 	"github.com/google/uuid"
@@ -54,6 +57,7 @@ var (
 	joinAddr     string
 	joinCAFinger string
 	leaveID      string
+	nodeWatch    bool
 )
 
 var nodeCmd = &cobra.Command{
@@ -155,8 +159,11 @@ var nodeLeaveCmd = &cobra.Command{
 var nodeListCmd = &cobra.Command{
 	Use:   "list",
 	Short: "List all nodes in the orca registry",
-	Long:  "Display all registered nodes and their state.",
+	Long:  "Display all registered nodes and their state. Use --watch to stream updates until Ctrl-C.",
 	RunE: func(cmd *cobra.Command, args []string) error {
+		if nodeWatch {
+			return watchNodes(cmd)
+		}
 		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
 		defer cancel()
 
@@ -185,11 +192,79 @@ var nodeListCmd = &cobra.Command{
 	},
 }
 
+func watchNodes(cmd *cobra.Command) error {
+	ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+	return watchNodesCtx(cmd, ctx)
+}
+
+func watchNodesCtx(cmd *cobra.Command, ctx context.Context) error {
+	db, closer, err := openDB()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	out := cmd.OutOrStdout()
+
+	if jsonOutput {
+		seen := make(map[string]string)
+		for snapshot := range store.NewNodeRepo(db).Watch(ctx) {
+			current := make(map[string]bool, len(snapshot))
+			for _, n := range snapshot {
+				current[n.ID] = true
+				compact, _ := json.Marshal(n)
+				key := string(compact)
+				if prev, ok := seen[n.ID]; !ok || prev != key {
+					event := "init"
+					if ok {
+						event = "update"
+					}
+					line, _ := json.Marshal(map[string]any{"event": event, "node": n})
+					fmt.Fprintln(out, string(line))
+					seen[n.ID] = key
+				}
+			}
+			for id := range seen {
+				if !current[id] {
+					line, _ := json.Marshal(map[string]any{"event": "delete", "id": id})
+					fmt.Fprintln(out, string(line))
+					delete(seen, id)
+				}
+			}
+		}
+		return nil
+	}
+
+	prevTable := ""
+	for snapshot := range store.NewNodeRepo(db).Watch(ctx) {
+		table := renderNodeTable(snapshot)
+		if table != prevTable {
+			fmt.Fprint(out, "\033[2J\033[H")
+			fmt.Fprint(out, table)
+			prevTable = table
+		}
+	}
+	return nil
+}
+
+func renderNodeTable(nodes []*model.Node) string {
+	if len(nodes) == 0 {
+		return "No nodes registered.\n"
+	}
+	out := fmt.Sprintf("%-36s  %-20s  %-22s  %-10s\n", "ID", "NAME", "ADDRESS", "STATE")
+	for _, n := range nodes {
+		out += fmt.Sprintf("%-36s  %-20s  %-22s  %-10s\n", n.ID, n.Name, n.Address, n.State)
+	}
+	return out
+}
+
 func init() {
 	nodeJoinCmd.Flags().StringVar(&joinName, "name", "", "node name (required)")
 	nodeJoinCmd.Flags().StringVar(&joinAddr, "addr", "", "node address (default localhost:8443)")
 	nodeJoinCmd.Flags().StringVar(&joinCAFinger, "ca-fingerprint", "", "pin CA cert SHA-256 (REQ-026); fails if on-disk CA doesn't match")
 	nodeLeaveCmd.Flags().StringVar(&leaveID, "id", "", "node id")
+	nodeListCmd.Flags().BoolVar(&nodeWatch, "watch", false, "stream nodes until Ctrl-C (table refresh or --json per-event)")
 
 	nodeCmd.AddCommand(nodeJoinCmd)
 	nodeCmd.AddCommand(nodeLeaveCmd)

internal/cli/watch_test.go

@@ -0,0 +1,287 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+func TestWatchJobs_JSONStreaming(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "orca.db")
+	db, err := store.Open(dbPath)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+
+	repo := store.NewJobRepo(db)
+	bgCtx := context.Background()
+	_ = repo.Insert(bgCtx, &model.Job{ID: "seed-job", Name: "seed", Spec: "t", Status: model.JobStatusPending})
+
+	t.Setenv("ORCA_DB", dbPath)
+
+	jsonOutput = true
+	t.Cleanup(func() { jsonOutput = false })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) })
+
+	ctx, cancel := context.WithCancel(bgCtx)
+
+	done := make(chan error, 1)
+	go func() { done <- watchJobsCtx(rootCmd, ctx) }()
+
+	// First yield is immediate (G-002); wait for it.
+	time.Sleep(100 * time.Millisecond)
+
+	_ = repo.Insert(bgCtx, &model.Job{ID: "watch-job", Name: "watch", Spec: "t", Status: model.JobStatusPending})
+
+	// Wait for at least one ticker interval (default 1s) to capture the change.
+	time.Sleep(1100 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watchJobsCtx did not return within 2s after cancel")
+	}
+
+	output := buf.String()
+	if !strings.Contains(output, `"event":"init"`) {
+		t.Errorf("expected init event, got: %s", output)
+	}
+	if !strings.Contains(output, "watch-job") {
+		t.Errorf("expected watch-job in output, got: %s", output)
+	}
+}
+
+func TestWatchJobs_TableRefresh(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "orca.db")
+	db, err := store.Open(dbPath)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+
+	repo := store.NewJobRepo(db)
+	bgCtx := context.Background()
+	_ = repo.Insert(bgCtx, &model.Job{ID: "seed-job", Name: "seed", Spec: "t", Status: model.JobStatusPending})
+
+	t.Setenv("ORCA_DB", dbPath)
+
+	jsonOutput = false
+	t.Cleanup(func() { jsonOutput = false })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) })
+
+	ctx, cancel := context.WithCancel(bgCtx)
+
+	done := make(chan error, 1)
+	go func() { done <- watchJobsCtx(rootCmd, ctx) }()
+
+	time.Sleep(100 * time.Millisecond)
+
+	_ = repo.Insert(bgCtx, &model.Job{ID: "table-job", Name: "table", Spec: "t", Status: model.JobStatusPending})
+
+	time.Sleep(1100 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watchJobsCtx did not return within 2s after cancel")
+	}
+
+	output := buf.String()
+	if !strings.Contains(output, "\033[2J\033[H") {
+		t.Errorf("expected clear-screen escape in table watch output, got: %s", output)
+	}
+	if !strings.Contains(output, "table-job") {
+		t.Errorf("expected table-job in output, got: %s", output)
+	}
+}
+
+func TestWatchNodes_JSONStreaming(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "orca.db")
+	db, err := store.Open(dbPath)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+
+	repo := store.NewNodeRepo(db)
+	bgCtx := context.Background()
+	_ = repo.Insert(bgCtx, &model.Node{
+		ID: "seed-node", Name: "seed", Address: "addr",
+		State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(),
+	})
+
+	t.Setenv("ORCA_DB", dbPath)
+
+	jsonOutput = true
+	t.Cleanup(func() { jsonOutput = false })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) })
+
+	ctx, cancel := context.WithCancel(bgCtx)
+
+	done := make(chan error, 1)
+	go func() { done <- watchNodesCtx(rootCmd, ctx) }()
+
+	time.Sleep(100 * time.Millisecond)
+
+	_ = repo.Insert(bgCtx, &model.Node{
+		ID: "watch-node", Name: "watch", Address: "addr2",
+		State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(),
+	})
+
+	time.Sleep(1100 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watchNodesCtx did not return within 2s after cancel")
+	}
+
+	output := buf.String()
+	initFound := false
+	watchNodeFound := false
+	for _, line := range strings.Split(output, "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		var event map[string]any
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			continue
+		}
+		if event["event"] == "init" {
+			initFound = true
+			if node, ok := event["node"].(map[string]any); ok {
+				if node["id"] == "watch-node" {
+					watchNodeFound = true
+				}
+			}
+		}
+	}
+	if !initFound {
+		t.Errorf("expected init event in JSON stream, got: %s", output)
+	}
+	if !watchNodeFound {
+		t.Errorf("expected watch-node in JSON stream, got: %s", output)
+	}
+}
+
+func TestWatchNodes_TableRefresh(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "orca.db")
+	db, err := store.Open(dbPath)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+
+	repo := store.NewNodeRepo(db)
+	bgCtx := context.Background()
+	_ = repo.Insert(bgCtx, &model.Node{
+		ID: "seed-node", Name: "seed", Address: "addr",
+		State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(),
+	})
+
+	t.Setenv("ORCA_DB", dbPath)
+
+	jsonOutput = false
+	t.Cleanup(func() { jsonOutput = false })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) })
+
+	ctx, cancel := context.WithCancel(bgCtx)
+
+	done := make(chan error, 1)
+	go func() { done <- watchNodesCtx(rootCmd, ctx) }()
+
+	time.Sleep(100 * time.Millisecond)
+
+	_ = repo.Insert(bgCtx, &model.Node{
+		ID: "table-node", Name: "table", Address: "addr2",
+		State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(),
+	})
+
+	time.Sleep(1100 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watchNodesCtx did not return within 2s after cancel")
+	}
+
+	output := buf.String()
+	if !strings.Contains(output, "\033[2J\033[H") {
+		t.Errorf("expected clear-screen escape in table watch output, got: %s", output)
+	}
+	if !strings.Contains(output, "table-node") {
+		t.Errorf("expected table-node in output, got: %s", output)
+	}
+}
+
+func TestRenderJobTable(t *testing.T) {
+	jobs := []*model.Job{
+		{ID: "j1", Name: "alpha", Status: "running", ExitCode: 0},
+		{ID: "j2", Name: "beta", Status: "done", ExitCode: 0},
+	}
+	out := renderJobTable(jobs)
+	if !strings.Contains(out, "j1") || !strings.Contains(out, "alpha") {
+		t.Errorf("renderJobTable missing job 1: %s", out)
+	}
+	if !strings.Contains(out, "j2") || !strings.Contains(out, "beta") {
+		t.Errorf("renderJobTable missing job 2: %s", out)
+	}
+}
+
+func TestRenderJobTableEmpty(t *testing.T) {
+	out := renderJobTable(nil)
+	if !strings.Contains(out, "No jobs") {
+		t.Errorf("expected empty message, got: %s", out)
+	}
+}
+
+func TestRenderNodeTable(t *testing.T) {
+	nodes := []*model.Node{
+		{ID: "n1", Name: "alpha", Address: "localhost:8443", State: "ready"},
+	}
+	out := renderNodeTable(nodes)
+	if !strings.Contains(out, "n1") || !strings.Contains(out, "alpha") {
+		t.Errorf("renderNodeTable missing node: %s", out)
+	}
+}
+
+func TestRenderNodeTableEmpty(t *testing.T) {
+	out := renderNodeTable(nil)
+	if !strings.Contains(out, "No nodes") {
+		t.Errorf("expected empty message, got: %s", out)
+	}
+}

internal/store/job_task_repo.go

@@ -6,11 +6,19 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"iter"
+	"log/slog"
 	"time"
 
 	"git.cloudinit.dev/coreci/orca/internal/model"
 )
 
+// watchInterval is the poll cadence used by JobRepo.Watch and NodeRepo.Watch.
+// It is an unexported package var (default 1s) so tests can override it to a
+// small value for deterministic assertions (D-043). Do not change it from
+// production code paths.
+var watchInterval = 1 * time.Second
+
 type JobRepo struct {
 	db *sql.DB
 }
@@ -59,6 +67,52 @@ func (r *JobRepo) List(ctx context.Context) ([]*model.Job, error) {
 	return jobs, rows.Err()
 }
 
+// Watch yields the full snapshot of jobs on a watchInterval ticker until ctx
+// is cancelled or the consumer stops pulling (yield returns false). It does
+// not spawn a goroutine; the polling loop runs inline in the caller's
+// goroutine via the range-over-func pull protocol (D-032).
+//
+// Each tick re-runs the List query and yields one []*model.Job snapshot
+// containing ALL rows for that tick (G-001). The first yield happens
+// immediately before the first ticker wait, so the consumer sees the initial
+// state with no watchInterval delay (G-002). Transient query/scan errors are
+// logged via slog.Default().Warn and the loop continues to the next tick
+// rather than terminating the stream (D-034 lite). The ticker is stopped and
+// rows are closed on every exit path (ctx.Done, yield==false, scan error).
+func (r *JobRepo) Watch(ctx context.Context) iter.Seq[[]*model.Job] {
+	return func(yield func([]*model.Job) bool) {
+		ticker := time.NewTicker(watchInterval)
+		defer ticker.Stop()
+		for {
+			rows, err := r.db.QueryContext(ctx,
+				`SELECT id, name, spec, status, exit_code, created_at, started_at, ended_at FROM jobs ORDER BY created_at DESC`)
+			if err != nil {
+				slog.Default().Warn("watch jobs: query failed", "error", err)
+				// fall through to the select to wait for the next tick
+			} else {
+				snapshot := make([]*model.Job, 0)
+				for rows.Next() {
+					j, scanErr := scanJob(rows)
+					if scanErr != nil {
+						slog.Default().Warn("watch jobs: scan failed", "error", scanErr)
+						continue
+					}
+					snapshot = append(snapshot, j)
+				}
+				rows.Close()
+				if !yield(snapshot) {
+					return // consumer stopped pulling
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}
+}
+
 func (r *JobRepo) UpdateStatus(ctx context.Context, id string, status model.JobStatus, exitCode int) error {
 	now := time.Now().UTC()
 	var startedAt, endedAt *time.Time

internal/store/job_task_repo_test.go

@@ -0,0 +1,201 @@
+package store
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/model"
+)
+
+func openJobTestDB(t *testing.T) (*JobRepo, func()) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "test.db")
+	db, err := Open(path)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	return NewJobRepo(db), func() { _ = db.Close() }
+}
+
+func insertJob(t *testing.T, repo *JobRepo, ctx context.Context, id, name string) {
+	t.Helper()
+	if err := repo.Insert(ctx, &model.Job{
+		ID:     id,
+		Name:   name,
+		Spec:   "test",
+		Status: model.JobStatusPending,
+	}); err != nil {
+		t.Fatalf("insert job %s: %v", id, err)
+	}
+}
+
+// withFastWatch sets watchInterval to a small value for deterministic tests and
+// restores the default (1s) on cleanup.
+func withFastWatch(t *testing.T, d time.Duration) {
+	t.Helper()
+	prev := watchInterval
+	watchInterval = d
+	t.Cleanup(func() { watchInterval = prev })
+}
+
+// TestJobRepoWatch_YieldsSnapshots verifies each yield is a complete tick
+// snapshot (G-001): the first snapshot contains only the first job, and a
+// later snapshot contains both jobs after a second insert.
+func TestJobRepoWatch_YieldsSnapshots(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openJobTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertJob(t, repo, ctx, "job-1", "alpha")
+
+	var snapshots [][]*model.Job
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for snap := range repo.Watch(ctx) {
+			snapshots = append(snapshots, snap)
+			if len(snapshots) >= 15 {
+				cancel()
+				return
+			}
+		}
+	}()
+
+	// Insert a second job after a short delay so a later tick observes it.
+	// Use a background context — the watch ctx may be cancelled by the
+	// goroutine above once it collects enough snapshots.
+	time.Sleep(100 * time.Millisecond)
+	insertJob(t, repo, context.Background(), "job-2", "beta")
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watch did not complete within 2s")
+	}
+
+	if len(snapshots) == 0 {
+		t.Fatal("expected at least one snapshot, got none")
+	}
+	// First snapshot must contain only the first job (G-001).
+	if len(snapshots[0]) != 1 || snapshots[0][0].ID != "job-1" {
+		t.Errorf("first snapshot = %+v, want only job-1", snapshots[0])
+	}
+	// At least one later snapshot must contain both jobs.
+	foundBoth := false
+	for _, snap := range snapshots[1:] {
+		ids := make(map[string]bool, len(snap))
+		for _, j := range snap {
+			ids[j.ID] = true
+		}
+		if ids["job-1"] && ids["job-2"] {
+			foundBoth = true
+			break
+		}
+	}
+	if !foundBoth {
+		t.Errorf("no snapshot contained both jobs; snapshots=%v", snapshots)
+	}
+}
+
+// TestJobRepoWatch_ImmediateFirstYield verifies G-002: the first snapshot
+// arrives before the first ticker wait, i.e. well under the watchInterval.
+func TestJobRepoWatch_ImmediateFirstYield(t *testing.T) {
+	withFastWatch(t, 200*time.Millisecond)
+	repo, cleanup := openJobTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertJob(t, repo, ctx, "job-immediate", "first")
+
+	start := time.Now()
+	var firstSnap []*model.Job
+	got := make(chan struct{})
+	go func() {
+		for snap := range repo.Watch(ctx) {
+			firstSnap = snap
+			close(got)
+			cancel()
+			return
+		}
+	}()
+
+	select {
+	case <-got:
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("first yield took >100ms; expected immediate (G-002)")
+	}
+
+	elapsed := time.Since(start)
+	if elapsed > 100*time.Millisecond {
+		t.Errorf("first yield took %v; expected immediate (G-002)", elapsed)
+	}
+	if len(firstSnap) != 1 || firstSnap[0].ID != "job-immediate" {
+		t.Errorf("first snapshot = %+v, want job-immediate", firstSnap)
+	}
+}
+
+// TestJobRepoWatch_StopsOnConsumerBreak verifies the yield==false path: the
+// range loop returns promptly when the consumer breaks after the first yield.
+func TestJobRepoWatch_StopsOnConsumerBreak(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openJobTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertJob(t, repo, ctx, "job-break", "break")
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range repo.Watch(ctx) {
+			break // stop pulling immediately after the first snapshot
+		}
+	}()
+
+	select {
+	case <-done:
+		// success: range returned
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("watch did not stop on consumer break within 500ms")
+	}
+}
+
+// TestJobRepoWatch_StopsOnCtxCancel verifies the loop exits promptly after
+// ctx is cancelled.
+func TestJobRepoWatch_StopsOnCtxCancel(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openJobTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	insertJob(t, repo, ctx, "job-cancel", "cancel")
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range repo.Watch(ctx) {
+			// drain until cancelled
+		}
+	}()
+
+	// Let at least one tick land, then cancel.
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+		// success
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("watch did not stop on ctx cancel within 500ms")
+	}
+}

internal/store/node_repo.go

@@ -6,6 +6,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"iter"
+	"log/slog"
 	"time"
 
 	"git.cloudinit.dev/coreci/orca/internal/model"
@@ -69,6 +71,40 @@ func (r *NodeRepo) List(ctx context.Context) ([]*model.Node, error) {
 	return nodes, rows.Err()
 }
 
+func (r *NodeRepo) Watch(ctx context.Context) iter.Seq[[]*model.Node] {
+	return func(yield func([]*model.Node) bool) {
+		ticker := time.NewTicker(watchInterval)
+		defer ticker.Stop()
+		for {
+			rows, err := r.db.QueryContext(ctx,
+				`SELECT id, name, address, state, joined_at, last_seen, metadata FROM nodes ORDER BY joined_at ASC`)
+			if err != nil {
+				slog.Default().Warn("watch nodes: query failed", "error", err)
+				// fall through to the select to wait for the next tick
+			} else {
+				snapshot := make([]*model.Node, 0)
+				for rows.Next() {
+					n, scanErr := scanNode(rows)
+					if scanErr != nil {
+						slog.Default().Warn("watch nodes: scan failed", "error", scanErr)
+						continue
+					}
+					snapshot = append(snapshot, n)
+				}
+				rows.Close()
+				if !yield(snapshot) {
+					return // consumer stopped pulling
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}
+}
+
 func (r *NodeRepo) UpdateState(ctx context.Context, id string, state model.NodeState) error {
 	res, err := r.db.ExecContext(ctx,
 		`UPDATE nodes SET state = ?, last_seen = ? WHERE id = ?`,

internal/store/node_repo_test.go

@@ -100,3 +100,159 @@ func TestNodeRepo_Delete(t *testing.T) {
 		t.Errorf("expected ErrNotFound, got %v", err)
 	}
 }
+
+func insertNode(t *testing.T, repo *NodeRepo, ctx context.Context, id, name string) {
+	t.Helper()
+	if err := repo.Insert(ctx, &model.Node{
+		ID:       id,
+		Name:     name,
+		Address:  "addr",
+		State:    model.NodeStateReady,
+		JoinedAt: time.Now().UTC(),
+		LastSeen: time.Now().UTC(),
+	}); err != nil {
+		t.Fatalf("insert node %s: %v", id, err)
+	}
+}
+
+func TestNodeRepoWatch_YieldsSnapshots(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertNode(t, repo, ctx, "node-1", "alpha")
+
+	var snapshots [][]*model.Node
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for snap := range repo.Watch(ctx) {
+			snapshots = append(snapshots, snap)
+			if len(snapshots) >= 15 {
+				cancel()
+				return
+			}
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+	insertNode(t, repo, context.Background(), "node-2", "beta")
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watch did not complete within 2s")
+	}
+
+	if len(snapshots) == 0 {
+		t.Fatal("expected at least one snapshot, got none")
+	}
+	if len(snapshots[0]) != 1 || snapshots[0][0].ID != "node-1" {
+		t.Errorf("first snapshot = %+v, want only node-1", snapshots[0])
+	}
+	foundBoth := false
+	for _, snap := range snapshots[1:] {
+		ids := make(map[string]bool, len(snap))
+		for _, n := range snap {
+			ids[n.ID] = true
+		}
+		if ids["node-1"] && ids["node-2"] {
+			foundBoth = true
+			break
+		}
+	}
+	if !foundBoth {
+		t.Errorf("no snapshot contained both nodes; snapshots=%v", snapshots)
+	}
+}
+
+func TestNodeRepoWatch_ImmediateFirstYield(t *testing.T) {
+	withFastWatch(t, 200*time.Millisecond)
+	repo, cleanup := openTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertNode(t, repo, ctx, "node-immediate", "first")
+
+	start := time.Now()
+	var firstSnap []*model.Node
+	got := make(chan struct{})
+	go func() {
+		for snap := range repo.Watch(ctx) {
+			firstSnap = snap
+			close(got)
+			cancel()
+			return
+		}
+	}()
+
+	select {
+	case <-got:
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("first yield took >100ms; expected immediate (G-002)")
+	}
+
+	elapsed := time.Since(start)
+	if elapsed > 100*time.Millisecond {
+		t.Errorf("first yield took %v; expected immediate (G-002)", elapsed)
+	}
+	if len(firstSnap) != 1 || firstSnap[0].ID != "node-immediate" {
+		t.Errorf("first snapshot = %+v, want node-immediate", firstSnap)
+	}
+}
+
+func TestNodeRepoWatch_StopsOnConsumerBreak(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	insertNode(t, repo, ctx, "node-break", "break")
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range repo.Watch(ctx) {
+			break
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("watch did not stop on consumer break within 500ms")
+	}
+}
+
+func TestNodeRepoWatch_StopsOnCtxCancel(t *testing.T) {
+	withFastWatch(t, 10*time.Millisecond)
+	repo, cleanup := openTestDB(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	insertNode(t, repo, ctx, "node-cancel", "cancel")
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range repo.Watch(ctx) {
+		}
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("watch did not stop on ctx cancel within 500ms")
+	}
+}