Persist v0.3 event pages and gate task admission

internal/configread/client.go

@@ -9,6 +9,7 @@ import (
 	"mime"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 
@@ -20,10 +21,9 @@ const dispatcherSecretHeader = "X-DISPATCHER-SECRET-KEY"
 const configReadPath = "/internal/v1/dispatcher"
 
 const (
-	resourceSIPConfig     = "sip_config"
-	resourceTaskConfig    = "task_config"
-	resourceTenantQuota   = "tenant_quota"
-	taskDiscoverySnapshot = "snapshot"
+	resourceSIPConfig   = "sip_config"
+	resourceTaskConfig  = "task_config"
+	resourceTenantQuota = "tenant_quota"
 )
 
 type Client struct {
@@ -37,7 +37,6 @@ type Snapshot struct {
 	TaskID                      string
 	TenantID                    string
 	TenantKey                   string
-	Discovery                   TaskDiscovery
 	SIPRevision                 int64
 	SIPSnapshotSHA256           string
 	SIPCellID                   string
@@ -62,20 +61,15 @@ type Snapshot struct {
 	ExpiresAt                   time.Time
 	SIP                         json.RawMessage
 	Task                        json.RawMessage
-	Tasks                       json.RawMessage
 	TenantQuota                 json.RawMessage
 }
 
 type TaskDiscovery struct {
 	DispatcherID string
-	Mode         string
-	Cursor       string
 	FromCursor   string
 	NextCursor   string
 	Tasks        []DiscoveredTask
-	Changes      []TaskDiscoveryChange
 	Body         json.RawMessage
-	Recovered    bool
 }
 
 type DiscoveredTask struct {
@@ -86,16 +80,6 @@ type DiscoveredTask struct {
 	TaskRevision int64  `json:"task_revision"`
 }
 
-type TaskDiscoveryChange struct {
-	Cursor       string `json:"cursor"`
-	Operation    string `json:"operation"`
-	TaskID       string `json:"task_id"`
-	TenantID     string `json:"tenant_id"`
-	TenantKey    string `json:"tenant_key"`
-	Status       string `json:"status"`
-	TaskRevision int64  `json:"task_revision"`
-}
-
 type HTTPError struct {
 	StatusCode int
 	Code       string
@@ -144,27 +128,6 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho
 		return Snapshot{}, errors.New("SIP configuration identity or revision does not match the request")
 	}
 
-	discovery, err := c.ReadTaskDiscovery(ctx, "")
-	if err != nil {
-		return Snapshot{}, err
-	}
-	if discovery.DispatcherID != c.dispatcherID || discovery.Mode != taskDiscoverySnapshot || discovery.Cursor == "" {
-		return Snapshot{}, errors.New("task discovery snapshot identity or mode does not match the request")
-	}
-	var discovered *DiscoveredTask
-	for i := range discovery.Tasks {
-		if discovery.Tasks[i].TaskID == taskID {
-			if discovered != nil {
-				return Snapshot{}, errors.New("task discovery contains duplicate task IDs")
-			}
-			discovered = &discovery.Tasks[i]
-		}
-	}
-	if discovered == nil || discovered.TenantID != tenantID {
-		return Snapshot{}, errors.New("task discovery does not bind the requested task to the requested tenant")
-	}
-	tasksBody := append(json.RawMessage(nil), discovery.Body...)
-
 	taskPath := configReadPath + "/task/" + url.PathEscape(taskID)
 	taskBody, err := c.getConfig(ctx, taskPath)
 	if err != nil {
@@ -177,10 +140,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho
 	if task.Resource != resourceTaskConfig || task.DispatcherID != c.dispatcherID || task.TaskID != taskID || task.TenantID != tenantID ||
 		task.TenantKey == "" || task.TaskRevision <= 0 || task.MaxConcurrentCalls < 0 || task.RingTimeoutMS <= 0 ||
 		task.MaxCallDurationMS <= 0 || task.Agent.AgentVersionID == "" || task.Agent.AgentVersionID != task.Agent.Config.AgentVersionID {
-		return Snapshot{}, errors.New("task configuration identity or limits do not match discovery")
-	}
-	if discovered.TenantKey != task.TenantKey || discovered.TaskRevision != task.TaskRevision || discovered.Status != task.Status {
-		return Snapshot{}, errors.New("task discovery and task configuration snapshots disagree")
+		return Snapshot{}, errors.New("task configuration identity or limits do not match the request")
 	}
 	authorizationExpiresAt, err := time.Parse(time.RFC3339, task.Agent.AuthorizationExpiresAt)
 	if err != nil {
@@ -205,7 +165,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho
 	}
 
 	return Snapshot{
-		TaskID: taskID, TenantID: tenantID, TenantKey: task.TenantKey, Discovery: discovery,
+		TaskID: taskID, TenantID: tenantID, TenantKey: task.TenantKey,
 		SIPRevision: sip.Revision, SIPSnapshotSHA256: sip.SnapshotSHA256,
 		SIPCellID: sip.Artifact.CellID, SIPArtifactRevision: sip.Artifact.Revision,
 		SIPArtifactConfigSHA256: sip.Artifact.ConfigSHA256,
@@ -216,8 +176,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho
 		AgentVersionID: task.Agent.AgentVersionID, AgentAuthorizationID: task.Agent.AuthorizationID,
 		AgentAuthorizationExpiresAt: authorizationExpiresAt, QuotaRevision: quota.QuotaRevision,
 		TenantMaxConcurrentCalls: int64(quota.MaxConcurrentCalls), QuotaValidUntil: quotaValidUntil,
-		DiscoveryCursor: discovery.Cursor,
-		SIP:             sipBody, Task: taskBody, Tasks: tasksBody, TenantQuota: quotaBody,
+		SIP: sipBody, Task: taskBody, TenantQuota: quotaBody,
 	}, nil
 }
 
@@ -262,27 +221,16 @@ func (c *Client) ReadTaskStatus(ctx context.Context, taskID, tenantID, tenantKey
 	}, nil
 }
 
-// ReadTaskDiscovery accepts exactly one complete snapshot or change response.
-// On HTTP 410 cursor_expired it fetches one full snapshot instead of advancing.
+// ReadTaskDiscovery reads one page of latest task states after a durable event ID.
+// SaaS guarantees page completeness and ordering; no per-item event ID is provided.
 func (c *Client) ReadTaskDiscovery(ctx context.Context, after string) (TaskDiscovery, error) {
-	return c.readTaskDiscovery(ctx, after, false)
-}
-
-func (c *Client) readTaskDiscovery(ctx context.Context, after string, recovered bool) (TaskDiscovery, error) {
-	path := configReadPath + "/tasks"
-	if after != "" {
-		path += "?" + url.Values{"after": {after}}.Encode()
+	from, err := parseEventCursor(after)
+	if err != nil {
+		return TaskDiscovery{}, fmt.Errorf("invalid task-discovery after cursor: %w", err)
 	}
+	path := configReadPath + "/tasks?" + url.Values{"after": {after}}.Encode()
 	body, err := c.getTaskDiscovery(ctx, path)
 	if err != nil {
-		var statusErr *HTTPError
-		if !recovered && after != "" && errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusGone && statusErr.Code == "cursor_expired" {
-			full, recoveryErr := c.readTaskDiscovery(ctx, "", true)
-			if recoveryErr != nil {
-				return TaskDiscovery{}, fmt.Errorf("recover expired task-discovery cursor with full snapshot: %w", recoveryErr)
-			}
-			return full, nil
-		}
 		return TaskDiscovery{}, err
 	}
 	var response taskDiscoveryResponse
@@ -292,32 +240,36 @@ func (c *Client) readTaskDiscovery(ctx context.Context, after string, recovered
 	if response.DispatcherID != c.dispatcherID {
 		return TaskDiscovery{}, errors.New("task-discovery dispatcher identity does not match the request")
 	}
-	result := TaskDiscovery{DispatcherID: response.DispatcherID, FromCursor: after, Body: append(json.RawMessage(nil), body...), Recovered: recovered}
+	to, err := parseEventCursor(response.NextCursor)
+	if err != nil {
+		return TaskDiscovery{}, fmt.Errorf("invalid task-discovery next cursor: %w", err)
+	}
+	if (len(response.Tasks) == 0 && to != from) || (len(response.Tasks) != 0 && to <= from) {
+		return TaskDiscovery{}, errors.New("task-discovery page cursor does not match returned tasks")
+	}
+	seen := make(map[string]struct{}, len(response.Tasks))
 	tenantByID, idByTenant := make(map[string]string), make(map[string]string)
-	if after == "" {
-		result.Mode, result.Cursor, result.Tasks = taskDiscoverySnapshot, response.Cursor, response.Tasks
-		seen := make(map[string]struct{}, len(result.Tasks))
-		for _, task := range result.Tasks {
-			if _, duplicate := seen[task.TaskID]; duplicate {
-				return TaskDiscovery{}, fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID)
-			}
-			seen[task.TaskID] = struct{}{}
-			if err := validateTenantBinding(tenantByID, idByTenant, task.TenantID, task.TenantKey); err != nil {
-				return TaskDiscovery{}, err
-			}
+	for _, task := range response.Tasks {
+		if _, duplicate := seen[task.TaskID]; duplicate {
+			return TaskDiscovery{}, fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID)
 		}
-	} else {
-		result.Mode, result.NextCursor, result.Changes = "changes", response.NextCursor, response.Changes
-		if (len(result.Changes) == 0 && result.NextCursor != after) || (len(result.Changes) != 0 && result.NextCursor == after) {
-			return TaskDiscovery{}, errors.New("task discovery change cursor does not match the complete response")
-		}
-		for _, change := range result.Changes {
-			if err := validateTenantBinding(tenantByID, idByTenant, change.TenantID, change.TenantKey); err != nil {
-				return TaskDiscovery{}, err
-			}
+		seen[task.TaskID] = struct{}{}
+		if err := validateTenantBinding(tenantByID, idByTenant, task.TenantID, task.TenantKey); err != nil {
+			return TaskDiscovery{}, err
 		}
 	}
-	return result, nil
+	return TaskDiscovery{
+		DispatcherID: response.DispatcherID, FromCursor: after, NextCursor: response.NextCursor,
+		Tasks: response.Tasks, Body: append(json.RawMessage(nil), body...),
+	}, nil
+}
+
+func parseEventCursor(cursor string) (uint64, error) {
+	value, err := strconv.ParseUint(cursor, 10, 64)
+	if err != nil || strconv.FormatUint(value, 10) != cursor {
+		return 0, errors.New("event cursor must be a canonical decimal uint64")
+	}
+	return value, nil
 }
 
 func validateTenantBinding(tenantByID, idByTenant map[string]string, tenantID, tenantKey string) error {
@@ -422,11 +374,9 @@ type sipConfigResponse struct {
 }
 
 type taskDiscoveryResponse struct {
-	DispatcherID string                `json:"dispatcher_id"`
-	Cursor       string                `json:"cursor"`
-	NextCursor   string                `json:"next_cursor"`
-	Tasks        []DiscoveredTask      `json:"tasks"`
-	Changes      []TaskDiscoveryChange `json:"changes"`
+	DispatcherID string           `json:"dispatcher_id"`
+	NextCursor   string           `json:"next_cursor"`
+	Tasks        []DiscoveredTask `json:"tasks"`
 }
 
 type taskConfigResponse struct {

internal/configread/client_test.go

@@ -22,7 +22,7 @@ const (
 	mockTestSecret   = "test-secret"
 )
 
-func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) {
+func TestClientReadTaskRequestsAndValidatesThreeConfigEndpoints(t *testing.T) {
 	fixtures := validConfigFixtures(t)
 	var got []string
 	mux := http.NewServeMux()
@@ -51,7 +51,7 @@ func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.Tasks) == 0 || len(snapshot.TenantQuota) == 0 {
+	if len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 {
 		t.Fatalf("incomplete config snapshot: %+v", snapshot)
 	}
 	if snapshot.TaskID != mockTaskID || snapshot.TenantID != mockTenantID || snapshot.TenantKey != mockTenantKey {
@@ -59,7 +59,6 @@ func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) {
 	}
 	want := []string{
 		"GET /internal/v1/dispatcher/sip",
-		"GET /internal/v1/dispatcher/tasks",
 		"GET /internal/v1/dispatcher/task/" + mockTaskID,
 		"GET /internal/v1/dispatcher/tenant/" + mockTenantID + "/quota",
 	}

internal/configread/discovery_v02_test.go

@@ -1,129 +0,0 @@
-package configread
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"net/http"
-	"net/http/httptest"
-	"reflect"
-	"strings"
-	"testing"
-)
-
-func TestTaskDiscoveryV02SnapshotSingleResponseAndNoQueue(t *testing.T) {
-	body := readConfigFixture(t, "task-discovery-snapshot-v0.2.json")
-	var requests []string
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		requests = append(requests, r.URL.RequestURI())
-		w.Header().Set("Content-Type", "application/json")
-		_, _ = w.Write(body)
-	}))
-	defer server.Close()
-	result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if result.Mode != "snapshot" || result.Cursor != "opaque-watermark-001" || len(result.Tasks) != 1 || string(result.Body) != string(body) {
-		t.Fatalf("snapshot = %+v", result)
-	}
-	if !reflect.DeepEqual(requests, []string{"/internal/v1/dispatcher/tasks"}) {
-		t.Fatalf("requests = %v, want one request", requests)
-	}
-}
-
-func TestTaskDiscoveryV02ChangesAndEmptyCursor(t *testing.T) {
-	for _, tc := range []struct {
-		name, file, after, next string
-		count                   int
-	}{
-		{"changes", "task-discovery-changes-v0.2.json", "opaque-watermark-001", "opaque-watermark-004", 3},
-		{"empty", "task-discovery-no-change-v0.2.json", "opaque-watermark-004", "opaque-watermark-004", 0},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			body := readConfigFixture(t, tc.file)
-			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				if r.URL.Query().Get("after") != tc.after || r.URL.Query().Has("page_token") {
-					t.Errorf("unexpected request %s", r.URL.String())
-				}
-				w.Header().Set("Content-Type", "application/json")
-				_, _ = w.Write(body)
-			}))
-			defer server.Close()
-			result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), tc.after)
-			if err != nil {
-				t.Fatal(err)
-			}
-			if result.Mode != "changes" || result.NextCursor != tc.next || len(result.Changes) != tc.count || result.FromCursor != tc.after {
-				t.Fatalf("changes = %+v", result)
-			}
-			if tc.count != 0 && (result.Changes[2].Operation != "removed" || result.Changes[2].TaskID != "task-old") {
-				t.Fatalf("removal = %+v", result.Changes[2])
-			}
-		})
-	}
-}
-
-func TestTaskDiscoveryV02ExpiredCursorFetchesOneFullSnapshot(t *testing.T) {
-	var requests []string
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		requests = append(requests, r.URL.RequestURI())
-		w.Header().Set("Content-Type", "application/json")
-		if r.URL.Query().Has("after") {
-			w.WriteHeader(http.StatusGone)
-			_, _ = fmt.Fprint(w, `{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"cursor_expired","message":"full snapshot required"}}`)
-			return
-		}
-		_, _ = w.Write(readConfigFixture(t, "task-discovery-snapshot-v0.2.json"))
-	}))
-	defer server.Close()
-	result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "stale")
-	if err != nil || !result.Recovered || result.Mode != "snapshot" || len(result.Tasks) != 1 {
-		t.Fatalf("recovery = %+v, %v", result, err)
-	}
-	if !reflect.DeepEqual(requests, []string{"/internal/v1/dispatcher/tasks?after=stale", "/internal/v1/dispatcher/tasks"}) {
-		t.Fatalf("requests = %v", requests)
-	}
-}
-
-func TestTaskDiscoveryV02RejectsObsoleteFieldsAndOversize(t *testing.T) {
-	for _, file := range []string{"task-discovery-invalid-queue-v0.2.json", "task-discovery-invalid-page-token-v0.2.json"} {
-		t.Run(file, func(t *testing.T) {
-			body := readConfigFixture(t, file)
-			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-				w.Header().Set("Content-Type", "application/json")
-				_, _ = w.Write(body)
-			}))
-			defer server.Close()
-			if _, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), ""); err == nil {
-				t.Fatal("accepted obsolete field")
-			}
-		})
-	}
-	for _, count := range []int{256, 257} {
-		t.Run(fmt.Sprint(count), func(t *testing.T) {
-			var base map[string]any
-			if err := json.Unmarshal(readConfigFixture(t, "task-discovery-snapshot-v0.2.json"), &base); err != nil {
-				t.Fatal(err)
-			}
-			tasks := make([]any, count)
-			for i := range tasks {
-				tasks[i] = map[string]any{"task_id": fmt.Sprintf("task-%03d", i), "tenant_id": mockTenantID, "tenant_key": mockTenantKey, "status": "running", "task_revision": 1}
-			}
-			base["tasks"] = tasks
-			body := marshalDiscoveryResponse(t, base)
-			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-				w.Header().Set("Content-Type", "application/json")
-				_, _ = w.Write(body)
-			}))
-			defer server.Close()
-			result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "")
-			if count == 256 && (err != nil || len(result.Tasks) != count) {
-				t.Fatalf("256 tasks: %d, %v", len(result.Tasks), err)
-			}
-			if count == 257 && (err == nil || !strings.Contains(err.Error(), "validate")) {
-				t.Fatalf("257 tasks: %v", err)
-			}
-		})
-	}
-}

internal/configread/discovery_v03_test.go

@@ -0,0 +1,134 @@
+package configread
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+const eventTestDispatcherID = "11111111-1111-4111-8111-111111111111"
+
+func eventTestClient(t *testing.T, server *httptest.Server) *Client {
+	t.Helper()
+	client, err := NewClient(server.URL, eventTestDispatcherID, mockTestSecret, server.Client())
+	if err != nil {
+		t.Fatal(err)
+	}
+	return client
+}
+
+func TestEventCursorPagesUseOneTaskListAndAllowRepeatedTaskIDs(t *testing.T) {
+	pages := map[string]string{
+		"0": "task-discovery-page-v0.3.json",
+		"2": "task-discovery-updated-v0.3.json",
+		"3": "task-discovery-removed-v0.3.json",
+		"4": "task-discovery-empty-v0.3.json",
+	}
+	var got []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		got = append(got, r.URL.RequestURI())
+		if r.Header.Get(dispatcherIDHeader) != eventTestDispatcherID || r.Header.Get(dispatcherSecretHeader) != mockTestSecret {
+			t.Error("Dispatcher credentials were not sent")
+		}
+		name := pages[r.URL.Query().Get("after")]
+		if name == "" || r.URL.Query().Has("page_token") {
+			t.Errorf("unexpected task request: %s", r.URL.RequestURI())
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write(readConfigFixture(t, name))
+	}))
+	defer server.Close()
+	client := eventTestClient(t, server)
+	cursor := "0"
+	var statuses []string
+	for range 4 {
+		page, err := client.ReadTaskDiscovery(context.Background(), cursor)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, task := range page.Tasks {
+			statuses = append(statuses, task.TaskID+":"+task.Status)
+		}
+		cursor = page.NextCursor
+	}
+	if cursor != "4" || !reflect.DeepEqual(statuses, []string{"a01:running", "a02:paused", "a01:paused", "a02:removed"}) {
+		t.Fatalf("statuses=%v cursor=%q", statuses, cursor)
+	}
+	want := []string{
+		"/internal/v1/dispatcher/tasks?after=0",
+		"/internal/v1/dispatcher/tasks?after=2",
+		"/internal/v1/dispatcher/tasks?after=3",
+		"/internal/v1/dispatcher/tasks?after=4",
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("queries=%v want=%v", got, want)
+	}
+}
+
+func TestEventCursorRejectsBrokenPagesWithoutFallback(t *testing.T) {
+	valid := string(readConfigFixture(t, "task-discovery-page-v0.3.json"))
+	var oversized map[string]any
+	if err := json.Unmarshal([]byte(valid), &oversized); err != nil {
+		t.Fatal(err)
+	}
+	rows := make([]any, 257)
+	for i := range rows {
+		rows[i] = oversized["tasks"].([]any)[0]
+	}
+	oversized["tasks"] = rows
+	oversizedBody, err := json.Marshal(oversized)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cases := map[string]string{
+		"nonprogress":      strings.Replace(valid, `"next_cursor": "2"`, `"next_cursor": "0"`, 1),
+		"duplicate":        strings.Replace(valid, `"task_id": "a02"`, `"task_id": "a01"`, 1),
+		"wrong dispatcher": strings.Replace(valid, eventTestDispatcherID, "44444444-4444-4444-8444-444444444444", 1),
+		"oversized page":   string(oversizedBody),
+		"changes":          string(readConfigFixture(t, "task-discovery-invalid-changes-v0.3.json")),
+		"cursor format":    string(readConfigFixture(t, "task-discovery-invalid-cursor-v0.3.json")),
+		"old contract":     string(readConfigFixture(t, "task-discovery-snapshot-v0.2.json")),
+	}
+	for name, body := range cases {
+		t.Run(name, func(t *testing.T) {
+			requests := 0
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				requests++
+				w.Header().Set("Content-Type", "application/json")
+				_, _ = w.Write([]byte(body))
+			}))
+			defer server.Close()
+			client := eventTestClient(t, server)
+			if _, err := client.ReadTaskDiscovery(context.Background(), "0"); err == nil {
+				t.Fatal("invalid task page was accepted")
+			}
+			if requests != 1 {
+				t.Fatalf("requests=%d, want one without fallback", requests)
+			}
+		})
+	}
+}
+
+func TestEventCursorRejectsExpiredStatusWithoutReset(t *testing.T) {
+	requests := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusGone)
+		_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"cursor_expired","message":"expired"}}`))
+	}))
+	defer server.Close()
+	client := eventTestClient(t, server)
+	if _, err := client.ReadTaskDiscovery(context.Background(), "1"); err == nil {
+		t.Fatal("HTTP 410 accepted")
+	}
+	if requests != 1 {
+		t.Fatalf("requests=%d, want one without reset to initial cursor", requests)
+	}
+}

internal/dispatcher/config_read.go

@@ -2,7 +2,6 @@ package dispatcher
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
@@ -49,6 +48,13 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C
 	if d.dispatcherID != "" && client.DispatcherID() != d.dispatcherID {
 		return fail(errors.New("configuration client dispatcher identity mismatch"))
 	}
+	assigned, err := d.store.LocalDiscoveredTaskForConfig(d.dispatcherID, taskID, tenantID)
+	if err != nil {
+		return fail(fmt.Errorf("authorize task configuration from durable discovery: %w", err))
+	}
+	if !d.now().UTC().Before(assigned.DiscoveryObservedAt.Add(taskDiscoveryTTL)) {
+		return fail(errors.New("task discovery is stale; new configuration admission is closed"))
+	}
 	snapshot, err := client.ReadTask(ctx, taskID, tenantID)
 	if err != nil {
 		return fail(err)
@@ -56,19 +62,28 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C
 	if err := verifier.VerifyAppliedSIPConfig(ctx, snapshot); err != nil {
 		return fail(fmt.Errorf("verify Agent-applied SIP config: %w", err))
 	}
+	assigned, err = d.store.LocalDiscoveredTaskForConfig(d.dispatcherID, taskID, tenantID)
+	if err != nil {
+		return fail(fmt.Errorf("recheck task discovery after configuration read: %w", err))
+	}
+	if snapshot.TenantKey != assigned.TenantKey || snapshot.TaskRevision != assigned.TaskRevision || snapshot.TaskStatus != assigned.Status {
+		return fail(errors.New("task configuration and durable discovery disagree"))
+	}
+	cursor, exists, err := d.store.LocalTaskDiscoveryCursor(d.dispatcherID)
+	if err != nil || !exists {
+		return fail(errors.New("durable task discovery cursor is unavailable"))
+	}
 	now := d.now().UTC()
 	quotaExpiry := minTime(now.Add(projectConfigTTL), snapshot.QuotaValidUntil)
 	taskExpiry := minTime(now.Add(projectConfigTTL), snapshot.AgentAuthorizationExpiresAt)
 	sipExpiry := now.Add(projectConfigTTL)
-	discoveryExpiry := now.Add(taskDiscoveryTTL)
+	discoveryExpiry := assigned.DiscoveryObservedAt.Add(taskDiscoveryTTL)
+	snapshot.DiscoveryCursor = cursor
 	snapshot.FetchedAt = now
 	snapshot.ExpiresAt = minTime(quotaExpiry, taskExpiry, sipExpiry, discoveryExpiry)
 	if !snapshot.ExpiresAt.After(now) {
 		return fail(errors.New("configuration response is expired; new admission is closed"))
 	}
-	if err := d.store.ApplyLocalTaskDiscovery(localTaskDiscoveryFromRead(snapshot.Discovery)); err != nil {
-		return fail(fmt.Errorf("persist complete task-discovery snapshot: %w", err))
-	}
 	bundle := store.LocalConfigBundle{
 		DispatcherID: d.dispatcherID, TenantID: snapshot.TenantID, TenantKey: snapshot.TenantKey, TaskID: snapshot.TaskID,
 		TenantQuotaScope: store.LocalTenantQuotaScope(d.dispatcherID, snapshot.TenantID),
@@ -76,7 +91,6 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C
 		TenantQuotaLimit: snapshot.TenantMaxConcurrentCalls, TaskQuotaLimit: snapshot.TaskMaxConcurrentCalls,
 		Snapshots: []store.LocalConfigSnapshotRecord{
 			{Resource: "sip", Revision: snapshot.SIPRevision, Version: strconv.FormatInt(snapshot.SIPRevision, 10), FetchedAt: now, ExpiresAt: sipExpiry, Body: snapshot.SIP},
-			{Resource: "tasks", Version: snapshot.DiscoveryCursor, FetchedAt: now, ExpiresAt: discoveryExpiry, Body: snapshot.Tasks},
 			{Resource: "task", TenantID: snapshot.TenantID, TaskID: snapshot.TaskID, Revision: snapshot.TaskRevision, Version: strconv.FormatInt(snapshot.TaskRevision, 10), FetchedAt: now, ExpiresAt: taskExpiry, Body: snapshot.Task},
 			{Resource: "tenant_quota", TenantID: snapshot.TenantID, Revision: snapshot.QuotaRevision, Version: strconv.FormatInt(snapshot.QuotaRevision, 10), FetchedAt: now, ExpiresAt: quotaExpiry, Body: snapshot.TenantQuota},
 		},
@@ -97,7 +111,6 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C
 		current.SIPCellID = snapshot.SIPCellID
 		current.SIPArtifactRevision = snapshot.SIPArtifactRevision
 		current.SIPArtifactConfigSHA256 = snapshot.SIPArtifactConfigSHA256
-		current.Tasks = append(current.Tasks[:0], snapshot.Tasks...)
 		current.DiscoveryCursor = snapshot.DiscoveryCursor
 		current.ExpiresAt = minTime(current.ExpiresAt, sharedExpiry)
 		if key.tenantID == snapshot.TenantID {
@@ -130,45 +143,13 @@ func (d *Dispatcher) ProjectConfigSnapshot(taskID, tenantID string) (configread.
 }
 
 func cloneConfigSnapshot(snapshot configread.Snapshot) configread.Snapshot {
-	snapshot.Discovery.Body = append(json.RawMessage(nil), snapshot.Discovery.Body...)
-	snapshot.Discovery.Tasks = append([]configread.DiscoveredTask(nil), snapshot.Discovery.Tasks...)
-	snapshot.Discovery.Changes = append([]configread.TaskDiscoveryChange(nil), snapshot.Discovery.Changes...)
 	snapshot.SIP = append(snapshot.SIP[:0:0], snapshot.SIP...)
 	snapshot.Task = append(snapshot.Task[:0:0], snapshot.Task...)
-	snapshot.Tasks = append(snapshot.Tasks[:0:0], snapshot.Tasks...)
 	snapshot.TenantQuota = append(snapshot.TenantQuota[:0:0], snapshot.TenantQuota...)
 	snapshot.TaskAllowedTrunkIDs = append([]string(nil), snapshot.TaskAllowedTrunkIDs...)
 	return snapshot
 }
 
-func localTaskDiscoveryFromRead(discovery configread.TaskDiscovery) store.LocalTaskDiscovery {
-	local := store.LocalTaskDiscovery{
-		DispatcherID: discovery.DispatcherID, Mode: discovery.Mode,
-		Cursor: discovery.Cursor, FromCursor: discovery.FromCursor,
-		NextCursor: discovery.NextCursor, Body: append([]byte(nil), discovery.Body...),
-	}
-	for _, task := range discovery.Tasks {
-		local.Tasks = append(local.Tasks, store.LocalDiscoveredTask{
-			TaskID: task.TaskID, TenantID: task.TenantID, TenantKey: task.TenantKey,
-			Status: task.Status, TaskRevision: task.TaskRevision,
-		})
-	}
-	for _, change := range discovery.Changes {
-		localChange := store.LocalTaskDiscoveryChange{
-			Cursor: change.Cursor, Operation: change.Operation, TaskID: change.TaskID,
-			TenantID: change.TenantID, TenantKey: change.TenantKey,
-		}
-		if change.Operation != "removed" {
-			localChange.Task = store.LocalDiscoveredTask{
-				TaskID: change.TaskID, TenantID: change.TenantID, TenantKey: change.TenantKey,
-				Status: change.Status, TaskRevision: change.TaskRevision,
-			}
-		}
-		local.Changes = append(local.Changes, localChange)
-	}
-	return local
-}
-
 func minTime(times ...time.Time) time.Time {
 	if len(times) == 0 {
 		return time.Time{}

internal/dispatcher/local_config_read_test.go

@@ -33,7 +33,7 @@ func TestDispatcherLoadsAndPersistsLocalConfigReadSnapshot(t *testing.T) {
 		t.Fatal("dispatcher did not retain an unexpired config snapshot")
 	}
 	if snapshot.TaskID != localTestTaskID || snapshot.TenantID != localTestTenantID ||
-		len(snapshot.SIP) == 0 || len(snapshot.Tasks) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 {
+		len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 {
 		t.Fatalf("incomplete configuration snapshot: %+v", snapshot)
 	}
 	persisted, err := st.LocalConfigSnapshot(localTestDispatcherID, "task", localTestTenantID, localTestTaskID)
@@ -139,6 +139,7 @@ func TestDispatcherSharesUpdatedTenantQuotaAcrossTasks(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	seedLocalEventDiscovery(t, st, now, "task-b")
 	if err := d.LoadProjectConfig(context.Background(), client, localTestSIPConfigVerifier(), "task-b", localTestTenantID); err != nil {
 		t.Fatal(err)
 	}

internal/dispatcher/local_v01.go

@@ -192,10 +192,10 @@ func (d *Dispatcher) AcceptLocalV01Command(ctx context.Context, routingKey strin
 				store.LocalTaskQuotaScope(d.dispatcherID, command.TenantID, projectConfig.TaskID),
 			},
 			Snapshot: store.LocalExecutionConfigSnapshot{
-				SchemaVersion: "execution-config-snapshot.v0.2", DispatcherID: d.dispatcherID,
+				SchemaVersion: "execution-config-snapshot.v0.3", DispatcherID: d.dispatcherID,
 				TenantID: projectConfig.TenantID, TenantKey: projectConfig.TenantKey, TaskID: projectConfig.TaskID, SelectedTrunkID: dial.TrunkID,
 				SIPRevision: projectConfig.SIPRevision, TaskRevision: projectConfig.TaskRevision, QuotaRevision: projectConfig.QuotaRevision,
-				SIP: projectConfig.SIP, Tasks: projectConfig.Tasks, Task: projectConfig.Task, TenantQuota: projectConfig.TenantQuota,
+				SIP: projectConfig.SIP, Task: projectConfig.Task, TenantQuota: projectConfig.TenantQuota,
 			},
 		}
 		capacityRejected := event

internal/dispatcher/local_v01_integration_test.go

@@ -35,7 +35,6 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) {
 	// only this isolated Mock grants a positive, explicit trunk limit.
 	fixtures := map[string][]byte{
 		"/internal/v1/dispatcher/sip":                                    localConfigFixture(t, "config-read-sip-v0.1.json"),
-		"/internal/v1/dispatcher/tasks":                                  readLocalFixture(t, "task-discovery-snapshot-v0.2.json"),
 		"/internal/v1/dispatcher/task/" + localTestTaskID:                readLocalFixture(t, "config-read-task-v0.1.json"),
 		"/internal/v1/dispatcher/tenant/" + localTestTenantID + "/quota": readLocalFixture(t, "config-read-tenant-quota-v0.1.json"),
 	}
@@ -50,6 +49,18 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) {
 			_, _ = fmt.Fprint(w, `{"schema_version":"config-read.v0.1","resource":"error","error":{"code":"unauthorized","message":"unauthorized"}}`)
 			return
 		}
+		if r.URL.Path == "/internal/v1/dispatcher/tasks" {
+			w.Header().Set("Content-Type", "application/json")
+			switch r.URL.Query().Get("after") {
+			case "0":
+				fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":"running","task_revision":2}],"next_cursor":"1"}`, localTestDispatcherID, localTestTaskID, localTestTenantID, localTestTenantKey)
+			case "1":
+				fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[],"next_cursor":"1"}`, localTestDispatcherID)
+			default:
+				w.WriteHeader(http.StatusBadRequest)
+			}
+			return
+		}
 		body, ok := fixtures[r.URL.Path]
 		if !ok {
 			w.WriteHeader(http.StatusNotFound)
@@ -140,6 +151,10 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	runtime := &LocalV01Runtime{dispatcher: d, client: configClient}
+	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
 	if err := d.LoadProjectConfig(context.Background(), configClient, localTestSIPConfigVerifier(), localTestTaskID, localTestTenantID); err != nil {
 		t.Fatal(err)
 	}
@@ -147,8 +162,9 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) {
 	gotRequests := append([]string(nil), requests...)
 	requestsMu.Unlock()
 	wantRequests := []string{
+		"GET /internal/v1/dispatcher/tasks?after=0",
+		"GET /internal/v1/dispatcher/tasks?after=1",
 		"GET /internal/v1/dispatcher/sip",
-		"GET /internal/v1/dispatcher/tasks",
 		"GET /internal/v1/dispatcher/task/" + localTestTaskID,
 		"GET /internal/v1/dispatcher/tenant/" + localTestTenantID + "/quota",
 	}

internal/dispatcher/local_v01_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -78,7 +79,7 @@ func TestAcceptLocalV01CommandPersistsReceiptAndRejectsExpiredCommand(t *testing
 		t.Fatal(err)
 	}
 	var bound store.LocalExecutionConfigSnapshot
-	if err := json.Unmarshal(executionBody, &bound); err != nil || bound.SelectedTrunkID != "trunk-mock" || bound.SchemaVersion != "execution-config-snapshot.v0.2" {
+	if err := json.Unmarshal(executionBody, &bound); err != nil || bound.SelectedTrunkID != "trunk-mock" || bound.SchemaVersion != "execution-config-snapshot.v0.3" {
 		t.Fatalf("accepted call did not bind its selected trunk: %+v err=%v", bound, err)
 	}
 	if err := st.MarkOutboxPublished(accepted[0].ID); err != nil {
@@ -517,7 +518,16 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store
 	t.Helper()
 	mux := http.NewServeMux()
 	mux.HandleFunc("/internal/v1/dispatcher/sip", localConfigResponse(string(localConfigFixture(t, "config-read-sip-v0.1.json"))))
-	mux.HandleFunc("/internal/v1/dispatcher/tasks", localConfigResponse(string(localConfigFixture(t, "task-discovery-snapshot-v0.2.json"))))
+	mux.HandleFunc("/internal/v1/dispatcher/tasks", func(w http.ResponseWriter, r *http.Request) {
+		after := r.URL.Query().Get("after")
+		if after == "" || after == "0" {
+			t.Errorf("expected durable event cursor, got %s", r.URL.RequestURI())
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":"%s","tasks":[],"next_cursor":%q}`, localTestDispatcherID, after)
+	})
 	mux.HandleFunc("/internal/v1/dispatcher/task/"+localTestTaskID, localConfigResponse(string(localConfigFixture(t, "config-read-task-v0.1.json"))))
 	mux.HandleFunc("/internal/v1/dispatcher/tenant/"+localTestTenantID+"/quota", localConfigResponse(string(localConfigFixture(t, "config-read-tenant-quota-v0.1.json"))))
 	server := httptest.NewServer(mux)
@@ -538,6 +548,7 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store
 		server.Close()
 		t.Fatal(err)
 	}
+	seedLocalEventDiscovery(t, st, now, localTestTaskID)
 	if err := d.LoadProjectConfig(context.Background(), client, localTestSIPConfigVerifier(), localTestTaskID, localTestTenantID); err != nil {
 		st.Close()
 		server.Close()
@@ -546,6 +557,42 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store
 	return d, st, server
 }
 
+func seedLocalEventDiscovery(t *testing.T, st *store.Store, now time.Time, taskIDs ...string) {
+	t.Helper()
+	var tasks []store.LocalDiscoveredTask
+	var wire []map[string]any
+	for _, taskID := range taskIDs {
+		tasks = append(tasks, store.LocalDiscoveredTask{TaskID: taskID, TenantID: localTestTenantID, TenantKey: localTestTenantKey, Status: "running", TaskRevision: 2})
+		wire = append(wire, map[string]any{"task_id": taskID, "tenant_id": localTestTenantID, "tenant_key": localTestTenantKey, "status": "running", "task_revision": 2})
+	}
+	after, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !exists {
+		after = "0"
+	}
+	position, err := strconv.ParseUint(after, 10, 64)
+	if err != nil || position == ^uint64(0) {
+		t.Fatalf("invalid test discovery cursor %q: %v", after, err)
+	}
+	next := strconv.FormatUint(position+1, 10)
+	body, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": localTestDispatcherID, "tasks": wire, "next_cursor": next})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{DispatcherID: localTestDispatcherID, FromCursor: after, NextCursor: next, Tasks: tasks, Body: body, ObservedAt: now}); err != nil {
+		t.Fatal(err)
+	}
+	empty, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": localTestDispatcherID, "tasks": []any{}, "next_cursor": next})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{DispatcherID: localTestDispatcherID, FromCursor: next, NextCursor: next, Body: empty, ObservedAt: now}); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func newLocalV01MultiTaskConfigServer(t *testing.T, taskIDs []string, quotaRevision, quotaLimit int64) *httptest.Server {
 	t.Helper()
 	if len(taskIDs) == 0 {

internal/dispatcher/task_runtime_v03_recovery_test.go

@@ -0,0 +1,143 @@
+package dispatcher
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"reflect"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestEventRuntimeResumesAtCommittedCursorAfterRestart(t *testing.T) {
+	var fail atomic.Bool
+	fail.Store(true)
+	var mu sync.Mutex
+	var requests []string
+	pages := map[string]string{
+		"0": "task-discovery-page-v0.3.json",
+		"2": "task-discovery-updated-v0.3.json",
+		"3": "task-discovery-removed-v0.3.json",
+		"4": "task-discovery-empty-v0.3.json",
+	}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		after := r.URL.Query().Get("after")
+		mu.Lock()
+		requests = append(requests, after)
+		mu.Unlock()
+		w.Header().Set("Content-Type", "application/json")
+		if r.URL.Path != "/internal/v1/dispatcher/tasks" || pages[after] == "" {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		if after == "2" && fail.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`))
+			return
+		}
+		_, _ = w.Write(eventRuntimeFixture(t, pages[after]))
+	}))
+	defer server.Close()
+	path := filepath.Join(t.TempDir(), "dispatcher.db")
+	runtime, st := newEventRuntimeAt(t, server, path)
+	if err := runtime.refreshDiscoveryState(context.Background()); err == nil {
+		t.Fatal("partial discovery accepted before restart")
+	}
+	cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID)
+	if err != nil || !exists || cursor != "2" {
+		t.Fatalf("checkpoint before restart=%q exists=%v err=%v", cursor, exists, err)
+	}
+	if err := st.Close(); err != nil {
+		t.Fatal(err)
+	}
+	fail.Store(false)
+	restarted, reopened := newEventRuntimeAt(t, server, path)
+	if err := restarted.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	mu.Lock()
+	got := append([]string(nil), requests...)
+	mu.Unlock()
+	if !reflect.DeepEqual(got, []string{"0", "2", "2", "3", "4"}) {
+		t.Fatalf("restart reset or skipped event cursor: requests=%v", got)
+	}
+	var ready int
+	if err := reopened.DB().QueryRow(`SELECT cursor,ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&cursor, &ready); err != nil || cursor != "4" || ready != 1 {
+		t.Fatalf("checkpoint after restart=%q ready=%d err=%v", cursor, ready, err)
+	}
+	assignments, err := reopened.LocalTaskAssignments(localTestDispatcherID)
+	if err != nil || len(assignments) != 2 || assignments[0].Status != "paused" || !assignments[1].Removed {
+		t.Fatalf("replayed assignments=%+v err=%v", assignments, err)
+	}
+}
+
+func TestEventRuntimeCatchesUpdateArrivingDuringPagination(t *testing.T) {
+	firstPageRequested := make(chan struct{})
+	updateCommitted := make(chan struct{})
+	var updated atomic.Bool
+	go func() {
+		<-firstPageRequested
+		updated.Store(true)
+		close(updateCommitted)
+	}()
+	var mu sync.Mutex
+	var requests []string
+	task := func(id, status string) string {
+		return fmt.Sprintf(`{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":%q,"task_revision":2}`, id, localTestTenantID, localTestTenantKey, status)
+	}
+	page := func(tasks, next string) []byte {
+		return []byte(fmt.Sprintf(`{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[%s],"next_cursor":%q}`, localTestDispatcherID, tasks, next))
+	}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		after := r.URL.Query().Get("after")
+		mu.Lock()
+		requests = append(requests, after)
+		mu.Unlock()
+		w.Header().Set("Content-Type", "application/json")
+		if r.URL.Path != "/internal/v1/dispatcher/tasks" {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		switch after {
+		case "0":
+			body := page(task(localTestTaskID, "running"), "1")
+			close(firstPageRequested)
+			<-updateCommitted
+			_, _ = w.Write(body)
+		case "1":
+			_, _ = w.Write(page(task("task-b", "running"), "2"))
+		case "2":
+			if !updated.Load() {
+				w.WriteHeader(http.StatusServiceUnavailable)
+				return
+			}
+			_, _ = w.Write(page(task(localTestTaskID, "paused"), "3"))
+		case "3":
+			_, _ = w.Write(page("", "3"))
+		default:
+			w.WriteHeader(http.StatusBadRequest)
+		}
+	}))
+	defer server.Close()
+	runtime, st := newEventRuntime(t, server)
+	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	mu.Lock()
+	got := append([]string(nil), requests...)
+	mu.Unlock()
+	if !reflect.DeepEqual(got, []string{"0", "1", "2", "3"}) {
+		t.Fatalf("concurrent update lost or pages skipped: %v", got)
+	}
+	assignments, err := st.LocalTaskAssignments(localTestDispatcherID)
+	if err != nil || len(assignments) != 2 {
+		t.Fatalf("incomplete assignments: %+v err=%v", assignments, err)
+	}
+	status := map[string]string{assignments[0].TaskID: assignments[0].Status, assignments[1].TaskID: assignments[1].Status}
+	if status[localTestTaskID] != "paused" || status["task-b"] != "running" {
+		t.Fatalf("update did not supersede earlier page: %+v", assignments)
+	}
+}

internal/dispatcher/task_runtime_v03_test.go

@@ -0,0 +1,125 @@
+package dispatcher
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"reflect"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+func eventRuntimeFixture(t *testing.T, name string) []byte {
+	t.Helper()
+	return bytes.ReplaceAll(localConfigFixture(t, name), []byte("11111111-1111-4111-8111-111111111111"), []byte(localTestDispatcherID))
+}
+
+func newEventRuntime(t *testing.T, server *httptest.Server) (*LocalV01Runtime, *store.Store) {
+	t.Helper()
+	return newEventRuntimeAt(t, server, filepath.Join(t.TempDir(), "dispatcher.db"))
+}
+
+func newEventRuntimeAt(t *testing.T, server *httptest.Server, path string) (*LocalV01Runtime, *store.Store) {
+	t.Helper()
+	st, err := store.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+	d, err := NewV3(localTestDispatcherID, st, nil, time.Now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client())
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &LocalV01Runtime{dispatcher: d, client: client}, st
+}
+
+func TestEventRuntimeBootsOnlyAfterAllPagesAndEmptyPage(t *testing.T) {
+	pages := map[string]string{"0": "task-discovery-page-v0.3.json", "2": "task-discovery-updated-v0.3.json", "3": "task-discovery-removed-v0.3.json", "4": "task-discovery-empty-v0.3.json"}
+	var afters []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		after := r.URL.Query().Get("after")
+		afters = append(afters, after)
+		if r.URL.Path != "/internal/v1/dispatcher/tasks" || r.URL.Query().Has("page_token") || pages[after] == "" {
+			t.Errorf("bad discovery request: %s", r.URL.RequestURI())
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write(eventRuntimeFixture(t, pages[after]))
+	}))
+	defer server.Close()
+	runtime, st := newEventRuntime(t, server)
+	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(afters, []string{"0", "2", "3", "4"}) {
+		t.Fatalf("pages read=%v", afters)
+	}
+	var ready int
+	if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 {
+		t.Fatalf("complete discovery ready=%d err=%v", ready, err)
+	}
+	assignments, err := st.LocalTaskAssignments(localTestDispatcherID)
+	if err != nil || len(assignments) != 2 || assignments[0].Status != "paused" || assignments[1].Status != "removed" {
+		t.Fatalf("assignments=%+v err=%v", assignments, err)
+	}
+	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if afters[len(afters)-1] != "4" {
+		t.Fatalf("poll did not resume durable cursor: %v", afters)
+	}
+}
+
+func TestEventRuntimeInterruptedPageResumesWithoutReset(t *testing.T) {
+	var fail atomic.Bool
+	fail.Store(true)
+	var afters []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		after := r.URL.Query().Get("after")
+		afters = append(afters, after)
+		w.Header().Set("Content-Type", "application/json")
+		if after == "2" && fail.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"test outage"}}`))
+			return
+		}
+		name := map[string]string{"0": "task-discovery-page-v0.3.json", "2": "task-discovery-updated-v0.3.json", "3": "task-discovery-removed-v0.3.json", "4": "task-discovery-empty-v0.3.json"}[after]
+		if name == "" {
+			t.Errorf("unexpected cursor %q", after)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		_, _ = w.Write(eventRuntimeFixture(t, name))
+	}))
+	defer server.Close()
+	runtime, st := newEventRuntime(t, server)
+	if err := runtime.refreshDiscoveryState(context.Background()); err == nil {
+		t.Fatal("partial page accepted")
+	}
+	cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID)
+	if err != nil || !exists || cursor != "2" {
+		t.Fatalf("partial cursor=%q exists=%v err=%v", cursor, exists, err)
+	}
+	var ready int
+	if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 {
+		t.Fatalf("partial response opened admission: ready=%d err=%v", ready, err)
+	}
+	fail.Store(false)
+	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(afters, []string{"0", "2", "2", "3", "4"}) {
+		t.Fatalf("retry must use persisted cursor, got %v", afters)
+	}
+}

internal/dispatcher/task_runtime_v3.go

@@ -153,17 +153,35 @@ func (r *LocalV01Runtime) refreshDiscoveryState(ctx context.Context) error {
 	if err != nil {
 		return r.failDiscovery(fmt.Errorf("read durable task-discovery cursor: %w", err))
 	}
+	if err := r.dispatcher.store.CloseLocalTaskDiscoveryAdmission(r.dispatcher.dispatcherID); err != nil {
+		return r.failDiscovery(fmt.Errorf("close task admission before discovery: %w", err))
+	}
 	if !exists {
-		cursor = ""
+		cursor = "0"
 	}
-	discovery, err := r.client.ReadTaskDiscovery(ctx, cursor)
-	if err != nil {
-		return r.failDiscovery(fmt.Errorf("read task discovery: %w", err))
+	pageCtx, cancel := context.WithTimeout(ctx, taskDiscoveryPollInterval)
+	defer cancel()
+	for {
+		page, err := r.client.ReadTaskDiscovery(pageCtx, cursor)
+		if err != nil {
+			return r.failDiscovery(fmt.Errorf("read task discovery after %s: %w", cursor, err))
+		}
+		tasks := make([]store.LocalDiscoveredTask, 0, len(page.Tasks))
+		for _, task := range page.Tasks {
+			tasks = append(tasks, store.LocalDiscoveredTask{TaskID: task.TaskID, TenantID: task.TenantID,
+				TenantKey: task.TenantKey, TaskRevision: task.TaskRevision, Status: task.Status})
+		}
+		if err := r.dispatcher.store.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{
+			DispatcherID: page.DispatcherID, FromCursor: cursor, NextCursor: page.NextCursor,
+			Tasks: tasks, Body: page.Body, ObservedAt: r.dispatcher.now().UTC(),
+		}); err != nil {
+			return r.failDiscovery(fmt.Errorf("persist task-discovery page after %s: %w", cursor, err))
+		}
+		if len(tasks) == 0 {
+			return nil
+		}
+		cursor = page.NextCursor
 	}
-	if err := r.dispatcher.store.ApplyLocalTaskDiscovery(localTaskDiscoveryFromRead(discovery)); err != nil {
-		return r.failDiscovery(fmt.Errorf("persist complete task-discovery response: %w", err))
-	}
-	return nil
 }
 
 func (r *LocalV01Runtime) failDiscovery(cause error) error {

internal/dispatcher/task_runtime_v3_test.go

@@ -2,6 +2,7 @@ package dispatcher
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"path/filepath"
@@ -13,64 +14,33 @@ import (
 	"git.ipao.vip/rogee/go-sip/internal/store"
 )
 
-func TestLocalV01RuntimeRefreshDiscoveryPersistsCompleteSnapshotAndCursor(t *testing.T) {
-	body := localConfigFixture(t, "task-discovery-snapshot-v0.2.json")
-	var afterValues []string
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodGet || r.URL.Path != "/internal/v1/dispatcher/tasks" {
-			t.Errorf("discovery request = %s %s", r.Method, r.URL.RequestURI())
-		}
-		afterValues = append(afterValues, r.URL.Query().Get("after"))
-		w.Header().Set("Content-Type", "application/json")
-		_, _ = w.Write(body)
-	}))
-	defer server.Close()
-	st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer st.Close()
-	d, err := NewV3(localTestDispatcherID, st, nil, time.Now)
-	if err != nil {
-		t.Fatal(err)
-	}
-	client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client())
-	if err != nil {
-		t.Fatal(err)
-	}
-	runtime := &LocalV01Runtime{dispatcher: d, client: client}
-	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	if len(afterValues) != 1 || afterValues[0] != "" {
-		t.Fatalf("initial discovery cursors = %v, want one full-snapshot request", afterValues)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID)
-	if err != nil || !exists || cursor == "" {
-		t.Fatalf("persisted cursor=%q exists=%v err=%v", cursor, exists, err)
-	}
-	assignments, err := st.LocalTaskAssignments(localTestDispatcherID)
-	if err != nil || len(assignments) != 1 || assignments[0].TaskID != localTestTaskID {
-		t.Fatalf("persisted assignments=%+v err=%v", assignments, err)
-	}
-}
-
 func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t *testing.T) {
 	var phase atomic.Int32
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		after := r.URL.Query().Get("after")
 		if r.URL.Path != "/internal/v1/dispatcher/tasks" {
 			t.Errorf("unexpected path: %s", r.URL.Path)
+			w.WriteHeader(http.StatusNotFound)
+			return
 		}
-		w.Header().Set("Content-Type", "application/json")
 		switch phase.Load() {
 		case 0:
-			_, _ = w.Write(localConfigFixture(t, "task-discovery-snapshot-v0.2.json"))
+			if after == "0" {
+				fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":"running","task_revision":2}],"next_cursor":"1"}`, localTestDispatcherID, localTestTaskID, localTestTenantID, localTestTenantKey)
+				return
+			}
 		case 1:
 			w.WriteHeader(http.StatusServiceUnavailable)
-			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`))
-		default:
-			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","next_cursor":"opaque-watermark-001","changes":[]}`))
+			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`))
+			return
 		}
+		if after != "1" {
+			t.Errorf("unexpected event cursor: %q", after)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[],"next_cursor":"1"}`, localTestDispatcherID)
 	}))
 	defer server.Close()
 	st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db"))
@@ -104,7 +74,7 @@ func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t *
 		t.Fatal("outage accepted as fresh discovery")
 	}
 	var ready int
-	if err := st.DB().QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 {
+	if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 {
 		t.Fatalf("discovery gate after outage=%d err=%v", ready, err)
 	}
 	select {
@@ -116,84 +86,8 @@ func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t *
 	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
 		t.Fatal(err)
 	}
-	if err := st.DB().QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 {
-		t.Fatalf("discovery gate after fresh no-change=%d err=%v", ready, err)
-	}
-}
-
-func TestLocalV01RuntimeExpiredCursorRecoversFullSnapshotAndRetiresTask(t *testing.T) {
-	var calls atomic.Int32
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/internal/v1/dispatcher/tasks" {
-			t.Errorf("unexpected path: %s", r.URL.Path)
-		}
-		w.Header().Set("Content-Type", "application/json")
-		switch calls.Add(1) {
-		case 1:
-			if r.URL.Query().Has("after") {
-				t.Error("initial read must be a full snapshot")
-			}
-			_, _ = w.Write(localConfigFixture(t, "task-discovery-snapshot-v0.2.json"))
-		case 2:
-			if r.URL.Query().Get("after") != "opaque-watermark-001" {
-				t.Errorf("incremental cursor = %s", r.URL.RawQuery)
-			}
-			w.WriteHeader(http.StatusGone)
-			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"cursor_expired","message":"mock expired"}}`))
-		case 3:
-			if r.URL.Query().Has("after") {
-				t.Errorf("recovery must be a full snapshot: %s", r.URL.RawQuery)
-			}
-			_, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","cursor":"opaque-watermark-002","tasks":[]}`))
-		default:
-			t.Error("unexpected fourth request")
-		}
-	}))
-	defer server.Close()
-	st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer st.Close()
-	d, err := NewV3(localTestDispatcherID, st, nil, time.Now)
-	if err != nil {
-		t.Fatal(err)
-	}
-	client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client())
-	if err != nil {
-		t.Fatal(err)
-	}
-	broker := &fakeQueueBroker{}
-	queues := newTaskQueueController(d, broker, nil)
-	runtime := &LocalV01Runtime{dispatcher: d, client: client, queues: queues}
-	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	assignment, err := st.LocalTaskAssignment(localTestDispatcherID, localTestTaskID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err := queues.StartTask(context.Background(), assignment); err != nil {
-		t.Fatal(err)
-	}
-	if err := runtime.refreshDiscoveryState(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	if err := runtime.reconcileTaskQueues(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID)
-	if err != nil || !exists || cursor != "opaque-watermark-002" || calls.Load() != 3 {
-		t.Fatalf("recovered cursor=%q exists=%v requests=%d err=%v", cursor, exists, calls.Load(), err)
-	}
-	assignment, err = st.LocalTaskAssignment(localTestDispatcherID, localTestTaskID)
-	if err != nil || !assignment.Removed || assignment.AdmissionState != "removed" {
-		t.Fatalf("retired assignment=%+v err=%v", assignment, err)
-	}
-	select {
-	case <-broker.consumers[0].done:
-	default:
-		t.Fatal("removed task remained subscribed")
+	if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 {
+		t.Fatalf("discovery gate after fresh empty page=%d err=%v", ready, err)
 	}
 }
 

internal/store/local_discovery_helpers_test.go

@@ -0,0 +1,96 @@
+package store
+
+import (
+	"encoding/json"
+	"path/filepath"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func openLocalDiscoveryTestStore(t *testing.T) *Store {
+	t.Helper()
+	st, err := Open(filepath.Join(t.TempDir(), "dispatcher.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+	return st
+}
+
+func seedLocalCommandAssignments(t *testing.T, st *Store, records ...LocalCommandRecord) {
+	t.Helper()
+	if len(records) == 0 {
+		return
+	}
+	dispatcherID := records[0].DispatcherID
+	tasks := make([]LocalDiscoveredTask, 0, len(records))
+	seen := make(map[string]struct{}, len(records))
+	for _, record := range records {
+		if record.DispatcherID != dispatcherID {
+			t.Fatalf("test commands use different Dispatcher IDs: %q and %q", dispatcherID, record.DispatcherID)
+		}
+		if _, exists := seen[record.TaskID]; exists {
+			continue
+		}
+		seen[record.TaskID] = struct{}{}
+		revision := int64(1)
+		if record.Admission != nil && record.Admission.Task.TaskRevision > 0 {
+			revision = record.Admission.Task.TaskRevision
+		}
+		tasks = append(tasks, localDiscoveredTask(dispatcherID, record.TaskID, record.TenantID, record.TenantKey, revision, "running"))
+	}
+	seedLocalEventTasks(t, st, dispatcherID, tasks...)
+}
+
+func seedLocalDiscovery(t *testing.T, st *Store) {
+	t.Helper()
+	seedLocalEventTasks(t, st, "d-1",
+		localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"),
+		localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 1, "running"),
+	)
+}
+
+func seedLocalEventTasks(t *testing.T, st *Store, dispatcherID string, tasks ...LocalDiscoveredTask) {
+	t.Helper()
+	cursor, exists, err := st.LocalTaskDiscoveryCursor(dispatcherID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !exists {
+		cursor = "0"
+	}
+	eventID, err := strconv.ParseUint(cursor, 10, 64)
+	if err != nil || eventID == ^uint64(0) {
+		t.Fatalf("invalid test event cursor %q: %v", cursor, err)
+	}
+	next := strconv.FormatUint(eventID+1, 10)
+	wire := make([]map[string]any, 0, len(tasks))
+	for _, task := range tasks {
+		wire = append(wire, map[string]any{"task_id": task.TaskID, "tenant_id": task.TenantID,
+			"tenant_key": task.TenantKey, "status": task.Status, "task_revision": task.TaskRevision})
+	}
+	body, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": dispatcherID,
+		"tasks": wire, "next_cursor": next})
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC()
+	if err := st.ApplyLocalTaskDiscoveryPage(LocalTaskDiscoveryPage{DispatcherID: dispatcherID, FromCursor: cursor,
+		NextCursor: next, Tasks: tasks, Body: body, ObservedAt: now}); err != nil {
+		t.Fatal(err)
+	}
+	empty, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": dispatcherID,
+		"tasks": []any{}, "next_cursor": next})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(LocalTaskDiscoveryPage{DispatcherID: dispatcherID, FromCursor: next,
+		NextCursor: next, Body: empty, ObservedAt: now}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func localDiscoveredTask(_ string, taskID, tenantID, tenantKey string, revision int64, status string) LocalDiscoveredTask {
+	return LocalDiscoveredTask{TaskID: taskID, TenantID: tenantID, TenantKey: tenantKey, TaskRevision: revision, Status: status}
+}

internal/store/local_origination.go

@@ -47,7 +47,7 @@ func (s *Store) LoadLocalOrigination(dispatcherID, executionID string) (LocalOri
 		JOIN local_v01_execution_configs c ON c.execution_id=t.execution_id
 		JOIN reservations r ON r.execution_id=t.execution_id
 		JOIN local_v01_task_assignments a ON a.dispatcher_id=? AND a.task_id=t.task_id
-		JOIN local_v02_task_discovery_state d ON d.dispatcher_id=a.dispatcher_id
+		JOIN local_v03_task_discovery_state d ON d.dispatcher_id=a.dispatcher_id
 		WHERE t.execution_id=?`, dispatcherID, executionID).Scan(&candidate.TenantID, &candidate.TenantKey, &candidate.TaskID, &candidate.CommandID, &candidate.Callee, &taskStatus,
 		&candidate.CommandBody, &inboxHash, &inboxStatus, &snapshotBody, &candidate.SnapshotSHA256, &reservationState, &admissionState, &removed, &ready)
 	if errors.Is(err, sql.ErrNoRows) {
@@ -75,7 +75,7 @@ func validateLocalOriginationCandidate(candidate *LocalOriginationCandidate, sna
 	if err := json.Unmarshal(snapshotBody, &candidate.Snapshot); err != nil {
 		return fmt.Errorf("decode bound execution snapshot: %w", err)
 	}
-	if candidate.Snapshot.SchemaVersion != "execution-config-snapshot.v0.2" || candidate.Snapshot.DispatcherID != candidate.DispatcherID ||
+	if candidate.Snapshot.SchemaVersion != "execution-config-snapshot.v0.3" || candidate.Snapshot.DispatcherID != candidate.DispatcherID ||
 		candidate.Snapshot.TaskID != candidate.TaskID || candidate.Snapshot.TenantID != candidate.TenantID || candidate.Snapshot.TenantKey != candidate.TenantKey ||
 		candidate.Snapshot.SelectedTrunkID == "" {
 		return fmt.Errorf("%w: execution snapshot identity or selected trunk mismatch", ErrLocalOriginationNotAuthorized)
@@ -104,7 +104,7 @@ func (s *Store) ClaimLocalOrigination(dispatcherID, executionID, agentID string,
 		SELECT t.execution_id,?,'issued','',? FROM tasks t WHERE t.execution_id=? AND t.status='reserved'
 		AND EXISTS (SELECT 1 FROM reservations r WHERE r.execution_id=t.execution_id AND r.state='held')
 		AND EXISTS (SELECT 1 FROM local_v01_task_assignments a WHERE a.dispatcher_id=? AND a.task_id=t.task_id AND a.tenant_id=t.tenant_id AND a.tenant_key=t.tenant_key AND a.removed=0 AND a.admission_state='running')
-		AND EXISTS (SELECT 1 FROM local_v02_task_discovery_state d WHERE d.dispatcher_id=? AND d.ready=1)
+		AND EXISTS (SELECT 1 FROM local_v03_task_discovery_state d WHERE d.dispatcher_id=? AND d.ready=1)
 		AND NOT EXISTS (SELECT 1 FROM controls c WHERE c.execution_id=t.execution_id AND c.action IN ('pause','drain','stop','hangup'))
 		AND NOT EXISTS (SELECT 1 FROM local_v02_origination_decisions old WHERE old.execution_id=t.execution_id)`,
 		dispatcherID, decidedAt.UTC().Format(time.RFC3339Nano), executionID, dispatcherID, dispatcherID)

internal/store/local_v01.go

@@ -49,7 +49,6 @@ type LocalExecutionConfigSnapshot struct {
 	TaskRevision    int64           `json:"task_revision"`
 	QuotaRevision   int64           `json:"quota_revision"`
 	SIP             json.RawMessage `json:"sip"`
-	Tasks           json.RawMessage `json:"tasks"`
 	Task            json.RawMessage `json:"task"`
 	TenantQuota     json.RawMessage `json:"tenant_quota"`
 }
@@ -191,9 +190,9 @@ func validateLocalAdmission(record LocalCommandRecord) ([]byte, error) {
 	if len(admission.QuotaScopes) != 2 || admission.QuotaScopes[0] != LocalTenantQuotaScope(admission.DispatcherID, task.TenantID) || admission.QuotaScopes[1] != LocalTaskQuotaScope(admission.DispatcherID, task.TenantID, task.TaskID) {
 		return nil, errors.New("local call admission quota scopes are invalid")
 	}
-	if snapshot.SchemaVersion != "execution-config-snapshot.v0.2" || snapshot.SelectedTrunkID == "" || snapshot.DispatcherID != admission.DispatcherID || snapshot.TenantID != task.TenantID || snapshot.TenantKey != task.TenantKey || snapshot.TaskID != task.TaskID ||
+	if snapshot.SchemaVersion != "execution-config-snapshot.v0.3" || snapshot.SelectedTrunkID == "" || snapshot.DispatcherID != admission.DispatcherID || snapshot.TenantID != task.TenantID || snapshot.TenantKey != task.TenantKey || snapshot.TaskID != task.TaskID ||
 		snapshot.TaskRevision != task.TaskRevision || snapshot.SIPRevision <= 0 || snapshot.QuotaRevision <= 0 ||
-		!validLocalJSONResponse(snapshot.SIP) || !validLocalJSONResponse(snapshot.Tasks) || !validLocalJSONResponse(snapshot.Task) || !validLocalJSONResponse(snapshot.TenantQuota) {
+		!validLocalJSONResponse(snapshot.SIP) || !validLocalJSONResponse(snapshot.Task) || !validLocalJSONResponse(snapshot.TenantQuota) {
 		return nil, errors.New("local call admission config snapshot identity or contents are invalid")
 	}
 	body, err := json.Marshal(snapshot)

internal/store/local_v01_admission_test.go

@@ -171,10 +171,10 @@ func localAdmissionCommand(commandID, executionID, taskID string, bundle LocalCo
 		responses[snapshot.Resource] = append(json.RawMessage(nil), snapshot.Body...)
 	}
 	config := LocalExecutionConfigSnapshot{
-		SchemaVersion: "execution-config-snapshot.v0.2", DispatcherID: dispatcherID,
+		SchemaVersion: "execution-config-snapshot.v0.3", DispatcherID: dispatcherID,
 		TenantID: tenantID, TenantKey: tenantKey, TaskID: taskID, SelectedTrunkID: "mock-trunk",
 		SIPRevision: bundle.Snapshots[0].Revision, TaskRevision: 1, QuotaRevision: 1,
-		SIP: responses["sip"], Tasks: responses["tasks"], Task: responses["task"], TenantQuota: responses["tenant_quota"],
+		SIP: responses["sip"], Task: responses["task"], TenantQuota: responses["tenant_quota"],
 	}
 	task := Task{
 		ExecutionID: executionID, TenantKey: tenantKey, TenantID: tenantID, TaskID: taskID,

internal/store/local_v01_config.go

@@ -54,10 +54,10 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error {
 	if bundle.DispatcherID == "" || bundle.TenantID == "" || bundle.TenantKey == "" || bundle.TaskID == "" ||
 		bundle.TenantQuotaLimit < 0 || bundle.TaskQuotaLimit < 0 ||
 		bundle.TenantQuotaScope != LocalTenantQuotaScope(bundle.DispatcherID, bundle.TenantID) ||
-		bundle.TaskQuotaScope != LocalTaskQuotaScope(bundle.DispatcherID, bundle.TenantID, bundle.TaskID) || len(bundle.Snapshots) != 4 {
+		bundle.TaskQuotaScope != LocalTaskQuotaScope(bundle.DispatcherID, bundle.TenantID, bundle.TaskID) || len(bundle.Snapshots) != 3 {
 		return errors.New("incomplete local configuration bundle")
 	}
-	seen := make(map[string]bool, 4)
+	seen := make(map[string]bool, 3)
 	for _, snapshot := range bundle.Snapshots {
 		if seen[snapshot.Resource] || !json.Valid(snapshot.Body) || snapshot.Version == "" ||
 			snapshot.FetchedAt.IsZero() || !snapshot.ExpiresAt.After(snapshot.FetchedAt) {
@@ -65,7 +65,7 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error {
 		}
 		seen[snapshot.Resource] = true
 		switch snapshot.Resource {
-		case "sip", "tasks":
+		case "sip":
 			if snapshot.TenantID != "" || snapshot.TaskID != "" {
 				return errors.New("shared configuration snapshot has task identity")
 			}
@@ -83,11 +83,8 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error {
 		if snapshot.Resource == "sip" && snapshot.Revision <= 0 {
 			return errors.New("SIP configuration revision must be positive")
 		}
-		if snapshot.Resource == "tasks" && snapshot.Revision != 0 {
-			return errors.New("task discovery uses an opaque version, not a numeric revision")
-		}
 	}
-	for _, resource := range []string{"sip", "tasks", "task", "tenant_quota"} {
+	for _, resource := range []string{"sip", "task", "tenant_quota"} {
 		if !seen[resource] {
 			return fmt.Errorf("missing local configuration resource %q", resource)
 		}
@@ -112,17 +109,11 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error {
 			WHERE dispatcher_id=? AND resource=? AND tenant_id=? AND task_id=?`,
 			bundle.DispatcherID, snapshot.Resource, snapshot.TenantID, snapshot.TaskID).Scan(&oldRevision, &oldVersion, &oldDigest)
 		if err == nil {
-			if snapshot.Resource == "tasks" {
-				if snapshot.Version == oldVersion && digestHex != oldDigest {
-					return ErrConfigRevisionConflict
-				}
-			} else {
-				if snapshot.Revision < oldRevision {
-					return ErrConfigRevisionRollback
-				}
-				if snapshot.Revision == oldRevision && (snapshot.Version != oldVersion || digestHex != oldDigest) {
-					return ErrConfigRevisionConflict
-				}
+			if snapshot.Revision < oldRevision {
+				return ErrConfigRevisionRollback
+			}
+			if snapshot.Revision == oldRevision && (snapshot.Version != oldVersion || digestHex != oldDigest) {
+				return ErrConfigRevisionConflict
 			}
 		} else if !errors.Is(err, sql.ErrNoRows) {
 			return err

internal/store/local_v01_config_test.go

@@ -42,7 +42,7 @@ func TestSaveLocalConfigBundlePersistsFullResponsesAndSharedQuota(t *testing.T)
 	if limit != 1 || reserved != 1 || unknown != 1 {
 		t.Fatalf("tenant quota after downgrade = limit:%d reserved:%d unknown:%d", limit, reserved, unknown)
 	}
-	if snapshot, err := st.LocalConfigSnapshot("d-1", "tenant_quota", "tenant-a", ""); err != nil || string(snapshot.Body) != string(second.Snapshots[3].Body) {
+	if snapshot, err := st.LocalConfigSnapshot("d-1", "tenant_quota", "tenant-a", ""); err != nil || string(snapshot.Body) != string(second.Snapshots[2].Body) {
 		t.Fatalf("shared tenant quota body=%q err=%v", snapshot.Body, err)
 	}
 }
@@ -70,7 +70,7 @@ func TestSaveLocalConfigBundleRejectsRevisionAndTenantBindingConflicts(t *testin
 		}
 	}
 	changedSameRevision := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 2, 2, 2, 3, 3, now.Add(time.Second))
-	changedSameRevision.Snapshots[2].Body = []byte(`{"resource":"task","revision":2,"different":true}`)
+	changedSameRevision.Snapshots[1].Body = []byte(`{"resource":"task","revision":2,"different":true}`)
 	if err := st.SaveLocalConfigBundle(changedSameRevision); !errors.Is(err, ErrConfigRevisionConflict) {
 		t.Fatalf("same revision content conflict error = %v, want ErrConfigRevisionConflict", err)
 	}
@@ -89,7 +89,6 @@ func localConfigBundle(dispatcherID, tenantID, tenantKey, taskID string, sipRevi
 		TenantQuotaLimit: tenantLimit, TaskQuotaLimit: taskLimit,
 		Snapshots: []LocalConfigSnapshotRecord{
 			{Resource: "sip", Revision: sipRevision, Version: fmt.Sprint(sipRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"sip","revision":%d}`, sipRevision))},
-			{Resource: "tasks", Version: fmt.Sprintf("snapshot-%s", taskID), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"tasks","snapshot_id":"snapshot-%s"}`, taskID))},
 			{Resource: "task", TenantID: tenantID, TaskID: taskID, Revision: taskRevision, Version: fmt.Sprint(taskRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"task","revision":%d,"task_id":%q}`, taskRevision, taskID))},
 			{Resource: "tenant_quota", TenantID: tenantID, Revision: quotaRevision, Version: fmt.Sprint(quotaRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"tenant_quota","revision":%d,"tenant_key":%q}`, quotaRevision, tenantKey))},
 		},

internal/store/local_v01_discovery.go

@@ -1,10 +1,7 @@
 package store
 
 import (
-	"crypto/sha256"
 	"database/sql"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -23,17 +20,6 @@ var (
 	ErrLocalTaskStopped             = errors.New("task admission is terminal")
 )
 
-type LocalTaskDiscovery struct {
-	DispatcherID string
-	Mode         string
-	Cursor       string
-	FromCursor   string
-	NextCursor   string
-	Body         []byte
-	Tasks        []LocalDiscoveredTask
-	Changes      []LocalTaskDiscoveryChange
-}
-
 type LocalDiscoveredTask struct {
 	TaskID       string
 	TenantID     string
@@ -42,15 +28,6 @@ type LocalDiscoveredTask struct {
 	TaskRevision int64
 }
 
-type LocalTaskDiscoveryChange struct {
-	Cursor    string
-	Operation string
-	TaskID    string
-	TenantID  string
-	TenantKey string
-	Task      LocalDiscoveredTask
-}
-
 type LocalTaskQueue struct {
 	Exchange   string
 	RoutingKey string
@@ -59,139 +36,17 @@ type LocalTaskQueue struct {
 }
 
 type LocalTaskAssignment struct {
-	DispatcherID   string
-	TaskID         string
-	TenantID       string
-	TenantKey      string
-	TaskRevision   int64
-	Status         string
-	AdmissionState string
-	Removed        bool
-	Queue          LocalTaskQueue
-	UpdatedAt      string
-}
-
-// ApplyLocalTaskDiscovery commits the complete response, assignments, and cursor
-// together. An invalid response cannot advance durable discovery state.
-func (s *Store) ApplyLocalTaskDiscovery(discovery LocalTaskDiscovery) error {
-	if err := validateLocalTaskDiscovery(discovery); err != nil {
-		return err
-	}
-	now := s.now().UTC().Format(time.RFC3339Nano)
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	tx, err := s.db.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
-
-	cursor := discovery.Cursor
-	if discovery.Mode == "changes" {
-		var currentCursor string
-		if err := tx.QueryRow(`SELECT cursor FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, discovery.DispatcherID).Scan(&currentCursor); err != nil {
-			if errors.Is(err, sql.ErrNoRows) {
-				return ErrLocalDiscoveryCursorMismatch
-			}
-			return err
-		}
-		if currentCursor != discovery.FromCursor {
-			return ErrLocalDiscoveryCursorMismatch
-		}
-		cursor = discovery.NextCursor
-	}
-
-	if discovery.Mode == "snapshot" {
-		seen := make(map[string]struct{}, len(discovery.Tasks))
-		for _, task := range discovery.Tasks {
-			if err := bindLocalTenant(tx, discovery.DispatcherID, task.TenantID, task.TenantKey, s.now()); err != nil {
-				return err
-			}
-			if err := applyLocalDiscoveredTask(tx, discovery.DispatcherID, task, true, now); err != nil {
-				return err
-			}
-			seen[task.TaskID] = struct{}{}
-		}
-		if err := markLocalTasksAbsent(tx, discovery.DispatcherID, seen, now); err != nil {
-			return err
-		}
-	} else {
-		for _, change := range discovery.Changes {
-			if err := applyLocalDiscoveryChange(tx, discovery.DispatcherID, change, now, s.now()); err != nil {
-				return err
-			}
-		}
-	}
-	var activeAssignments int
-	if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, discovery.DispatcherID).Scan(&activeAssignments); err != nil {
-		return err
-	}
-	if activeAssignments > 256 {
-		return ErrLocalDiscoveryTaskLimit
-	}
-	digest := sha256.Sum256(discovery.Body)
-	if _, err := tx.Exec(`INSERT INTO local_v02_task_discovery_response(dispatcher_id,content_sha256,body,stored_at)
-		VALUES(?,?,?,?) ON CONFLICT(dispatcher_id) DO UPDATE SET content_sha256=excluded.content_sha256,
-		body=excluded.body,stored_at=excluded.stored_at`, discovery.DispatcherID, hex.EncodeToString(digest[:]), discovery.Body, now); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(`INSERT INTO local_v02_task_discovery_state(dispatcher_id,cursor,last_mode,ready,updated_at)
-		VALUES(?,?,?,1,?) ON CONFLICT(dispatcher_id) DO UPDATE SET cursor=excluded.cursor,
-		last_mode=excluded.last_mode,ready=1,updated_at=excluded.updated_at`, discovery.DispatcherID, cursor, discovery.Mode, now); err != nil {
-		return err
-	}
-	return tx.Commit()
-}
-
-func validateLocalTaskDiscovery(discovery LocalTaskDiscovery) error {
-	if discovery.DispatcherID == "" || len(discovery.Body) == 0 || !json.Valid(discovery.Body) {
-		return errors.New("task discovery requires a Dispatcher ID and one valid response body")
-	}
-	switch discovery.Mode {
-	case "snapshot":
-		if discovery.Cursor == "" || discovery.FromCursor != "" || discovery.NextCursor != "" || len(discovery.Changes) != 0 || len(discovery.Tasks) > 256 {
-			return errors.New("invalid task-discovery snapshot window")
-		}
-		seen := make(map[string]struct{}, len(discovery.Tasks))
-		for _, task := range discovery.Tasks {
-			if err := validateLocalDiscoveredTask(discovery.DispatcherID, task); err != nil {
-				return err
-			}
-			if _, exists := seen[task.TaskID]; exists {
-				return fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID)
-			}
-			seen[task.TaskID] = struct{}{}
-		}
-	case "changes":
-		if discovery.FromCursor == "" || discovery.NextCursor == "" || discovery.Cursor != "" || len(discovery.Tasks) != 0 || len(discovery.Changes) > 256 ||
-			(len(discovery.Changes) == 0 && discovery.NextCursor != discovery.FromCursor) ||
-			(len(discovery.Changes) != 0 && discovery.NextCursor == discovery.FromCursor) {
-			return errors.New("invalid task-discovery change window")
-		}
-		for _, change := range discovery.Changes {
-			if change.Cursor == "" {
-				return errors.New("task-discovery change cursor is required")
-			}
-			switch change.Operation {
-			case "assigned", "updated":
-				if err := validateLocalDiscoveredTask(discovery.DispatcherID, change.Task); err != nil {
-					return err
-				}
-				if change.TaskID != "" && change.TaskID != change.Task.TaskID {
-					return errors.New("task-discovery change task ID conflicts with its task payload")
-				}
-			case "removed":
-				if !validLocalTaskID(change.TaskID) || change.TenantID == "" || change.TenantKey == "" || len([]byte(change.TenantKey)) > 196 {
-					return errors.New("invalid task-discovery removal tombstone")
-				}
-			default:
-				return fmt.Errorf("unsupported task-discovery operation %q", change.Operation)
-			}
-		}
-	default:
-		return fmt.Errorf("unsupported task-discovery mode %q", discovery.Mode)
-	}
-	return nil
+	DispatcherID        string
+	TaskID              string
+	TenantID            string
+	TenantKey           string
+	TaskRevision        int64
+	Status              string
+	AdmissionState      string
+	Removed             bool
+	Queue               LocalTaskQueue
+	UpdatedAt           string
+	DiscoveryObservedAt time.Time
 }
 
 func validateLocalDiscoveredTask(dispatcherID string, task LocalDiscoveredTask) error {
@@ -199,7 +54,7 @@ func validateLocalDiscoveredTask(dispatcherID string, task LocalDiscoveredTask)
 		return errors.New("incomplete task-discovery assignment")
 	}
 	switch task.Status {
-	case "running", "paused", "stopped", "finished":
+	case "running", "paused", "stopped", "finished", "removed":
 	default:
 		return fmt.Errorf("unsupported discovered task status %q", task.Status)
 	}
@@ -224,7 +79,7 @@ func derivedLocalTaskQueue(dispatcherID, taskID string) LocalTaskQueue {
 		QueueName: fmt.Sprintf("agent-call.d.%s.task.%s.v3", dispatcherID, taskID)}
 }
 
-func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscoveredTask, allowCreate bool, now string) error {
+func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscoveredTask, now string) error {
 	queue := derivedLocalTaskQueue(dispatcherID, task.TaskID)
 	var old LocalTaskAssignment
 	var removed int
@@ -233,12 +88,13 @@ func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscove
 		dispatcherID, task.TaskID).Scan(&old.DispatcherID, &old.TaskID, &old.TenantID, &old.TenantKey, &old.TaskRevision,
 		&old.Status, &old.AdmissionState, &removed, &old.Queue.Exchange, &old.Queue.RoutingKey, &old.Queue.BindingKey, &old.Queue.QueueName, &old.UpdatedAt)
 	if errors.Is(err, sql.ErrNoRows) {
-		if !allowCreate {
-			return ErrLocalDiscoveryTaskMissing
+		removed := 0
+		if task.Status == "removed" {
+			removed = 1
 		}
 		_, err = tx.Exec(`INSERT INTO local_v01_task_assignments(dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed,
-			queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,?,?,?,0,?,?,?,?,?)`,
-			dispatcherID, task.TaskID, task.TenantID, task.TenantKey, task.TaskRevision, task.Status, task.Status,
+			queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`,
+			dispatcherID, task.TaskID, task.TenantID, task.TenantKey, task.TaskRevision, task.Status, task.Status, removed,
 			queue.Exchange, queue.RoutingKey, queue.BindingKey, queue.QueueName, now)
 		return err
 	}
@@ -246,24 +102,31 @@ func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscove
 		return err
 	}
 	old.Removed = removed != 0
-	if old.Removed {
-		return ErrLocalTaskRemoved
-	}
 	if old.TenantID != task.TenantID || old.TenantKey != task.TenantKey {
 		return ErrTenantBindingConflict
 	}
 	if task.TaskRevision < old.TaskRevision {
 		return ErrConfigRevisionRollback
 	}
-	if task.TaskRevision == old.TaskRevision {
-		if old.Status != task.Status || old.Queue != queue {
-			return ErrConfigRevisionConflict
+	if old.Removed {
+		if task.Status != "removed" {
+			return ErrLocalTaskRemoved
 		}
 		return nil
 	}
+	if old.Queue != queue {
+		return ErrConfigRevisionConflict
+	}
+	if task.TaskRevision == old.TaskRevision && task.Status == old.Status {
+		return nil
+	}
 	state := mergeLocalTaskState(old.AdmissionState, task.Status)
-	_, err = tx.Exec(`UPDATE local_v01_task_assignments SET task_revision=?,saas_status=?,admission_state=?,queue_exchange=?,routing_key=?,binding_key=?,queue_name=?,updated_at=?
-		WHERE dispatcher_id=? AND task_id=?`, task.TaskRevision, task.Status, state, queue.Exchange, queue.RoutingKey,
+	removed = 0
+	if task.Status == "removed" {
+		state, removed = "removed", 1
+	}
+	_, err = tx.Exec(`UPDATE local_v01_task_assignments SET task_revision=?,saas_status=?,admission_state=?,removed=?,queue_exchange=?,routing_key=?,binding_key=?,queue_name=?,updated_at=?
+		WHERE dispatcher_id=? AND task_id=?`, task.TaskRevision, task.Status, state, removed, queue.Exchange, queue.RoutingKey,
 		queue.BindingKey, queue.QueueName, now, dispatcherID, task.TaskID)
 	return err
 }
@@ -288,94 +151,30 @@ func mergeLocalTaskState(current, incoming string) string {
 	}
 }
 
-func applyLocalDiscoveryChange(tx *sql.Tx, dispatcherID string, change LocalTaskDiscoveryChange, now string, createdAt time.Time) error {
-	switch change.Operation {
-	case "assigned":
-		if err := bindLocalTenant(tx, dispatcherID, change.Task.TenantID, change.Task.TenantKey, createdAt); err != nil {
-			return err
-		}
-		return applyLocalDiscoveredTask(tx, dispatcherID, change.Task, true, now)
-	case "updated":
-		if err := bindLocalTenant(tx, dispatcherID, change.Task.TenantID, change.Task.TenantKey, createdAt); err != nil {
-			return err
-		}
-		return applyLocalDiscoveredTask(tx, dispatcherID, change.Task, false, now)
-	case "removed":
-		if err := bindLocalTenant(tx, dispatcherID, change.TenantID, change.TenantKey, createdAt); err != nil {
-			return err
-		}
-		return removeLocalTaskTx(tx, dispatcherID, change, now)
-	default:
-		return fmt.Errorf("unsupported task-discovery operation %q", change.Operation)
-	}
-}
-
-func removeLocalTaskTx(tx *sql.Tx, dispatcherID string, change LocalTaskDiscoveryChange, now string) error {
-	var tenantID, tenantKey string
-	err := tx.QueryRow(`SELECT tenant_id,tenant_key FROM local_v01_task_assignments WHERE dispatcher_id=? AND task_id=?`, dispatcherID, change.TaskID).Scan(&tenantID, &tenantKey)
-	if errors.Is(err, sql.ErrNoRows) {
-		queue := derivedLocalTaskQueue(dispatcherID, change.TaskID)
-		_, err = tx.Exec(`INSERT INTO local_v01_task_assignments(dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed,
-			queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,0,'removed','removed',1,?,?,?,?,?)`,
-			dispatcherID, change.TaskID, change.TenantID, change.TenantKey, queue.Exchange, queue.RoutingKey, queue.BindingKey, queue.QueueName, now)
-		return err
-	}
-	if err != nil {
-		return err
-	}
-	if tenantID != change.TenantID || tenantKey != change.TenantKey {
-		return ErrTenantBindingConflict
-	}
-	_, err = tx.Exec(`UPDATE local_v01_task_assignments SET saas_status='removed',admission_state='removed',removed=1,updated_at=?
-		WHERE dispatcher_id=? AND task_id=?`, now, dispatcherID, change.TaskID)
-	return err
-}
-
-func markLocalTasksAbsent(tx *sql.Tx, dispatcherID string, present map[string]struct{}, now string) error {
-	rows, err := tx.Query(`SELECT task_id FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, dispatcherID)
-	if err != nil {
-		return err
-	}
-	var absent []string
-	for rows.Next() {
-		var taskID string
-		if err := rows.Scan(&taskID); err != nil {
-			_ = rows.Close()
-			return err
-		}
-		if _, exists := present[taskID]; !exists {
-			absent = append(absent, taskID)
-		}
-	}
-	if err := rows.Err(); err != nil {
-		_ = rows.Close()
-		return err
-	}
-	if err := rows.Close(); err != nil {
-		return err
-	}
-	for _, taskID := range absent {
-		if _, err := tx.Exec(`UPDATE local_v01_task_assignments SET saas_status='removed',admission_state='removed',removed=1,updated_at=? WHERE dispatcher_id=? AND task_id=?`, now, dispatcherID, taskID); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 // CloseLocalTaskDiscoveryAdmission rejects new work without losing the cursor
 // or changing an authoritative task pause/stop state.
 func (s *Store) CloseLocalTaskDiscoveryAdmission(dispatcherID string) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	_, err := s.db.Exec(`UPDATE local_v02_task_discovery_state SET ready=0 WHERE dispatcher_id=?`, dispatcherID)
+	_, err := s.db.Exec(`UPDATE local_v03_task_discovery_state SET ready=0 WHERE dispatcher_id=?`, dispatcherID)
 	return err
 }
 
 func (s *Store) LocalTaskDiscoveryCursor(dispatcherID string) (cursor string, exists bool, err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	err = s.db.QueryRow(`SELECT cursor FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&cursor)
+	err = s.db.QueryRow(`SELECT cursor FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&cursor)
 	if errors.Is(err, sql.ErrNoRows) {
+		var legacy, assignments int
+		if err := s.db.QueryRow(`SELECT COUNT(*) FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&legacy); err != nil {
+			return "", false, err
+		}
+		if err := s.db.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=?`, dispatcherID).Scan(&assignments); err != nil {
+			return "", false, err
+		}
+		if legacy != 0 || assignments != 0 {
+			return "", false, fmt.Errorf("%w: existing discovery state requires controlled drain before v0.3", ErrLocalDiscoveryUnavailable)
+		}
 		return "", false, nil
 	}
 	if err != nil {
@@ -429,7 +228,7 @@ func scanLocalTaskAssignment(row interface{ Scan(...any) error }) (LocalTaskAssi
 
 func requireLocalTaskRunningTx(tx *sql.Tx, dispatcherID, tenantID, tenantKey, taskID string) error {
 	var ready int
-	if err := tx.QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready); err != nil {
+	if err := tx.QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return ErrLocalDiscoveryUnavailable
 		}

internal/store/local_v01_discovery_test.go

@@ -1,292 +0,0 @@
-package store
-
-import (
-	"crypto/sha256"
-	"database/sql"
-	"encoding/hex"
-	"errors"
-	"fmt"
-	"path/filepath"
-	"testing"
-	"time"
-
-	"git.ipao.vip/rogee/go-sip/internal/mq"
-)
-
-func TestApplyLocalTaskDiscoveryPersistsSingleResponseAndDerivedQueue(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	body := []byte(`{"cursor":"opaque-cursor-1","tasks":[{"task_id":"task-a"}]}`)
-	if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "opaque-cursor-1", body,
-		localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"),
-		localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 4, "paused"),
-	)); err != nil {
-		t.Fatal(err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || !exists || cursor != "opaque-cursor-1" {
-		t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err)
-	}
-	assignments, err := st.LocalTaskAssignments("d-1")
-	if err != nil || len(assignments) != 2 {
-		t.Fatalf("assignments=%+v err=%v", assignments, err)
-	}
-	if assignments[0].TaskID != "task-a" || assignments[0].AdmissionState != "running" || assignments[1].AdmissionState != "paused" {
-		t.Fatalf("initial task states = %+v", assignments)
-	}
-	if assignments[0].Queue.Exchange != mq.CommandsExchangeV3 || assignments[0].Queue.RoutingKey != "d.d-1.task.task-a.in" || assignments[0].Queue.BindingKey != "d.d-1.task.task-a.in" || assignments[0].Queue.QueueName != "agent-call.d.d-1.task.task-a.v3" {
-		t.Fatalf("derived queue = %+v", assignments[0].Queue)
-	}
-	var saved []byte
-	var digest string
-	if err := st.db.QueryRow(`SELECT body,content_sha256 FROM local_v02_task_discovery_response WHERE dispatcher_id=?`, "d-1").Scan(&saved, &digest); err != nil {
-		t.Fatal(err)
-	}
-	hash := sha256.Sum256(body)
-	if string(saved) != string(body) || digest != hex.EncodeToString(hash[:]) {
-		t.Fatalf("stored discovery response mismatch: %s %s", saved, digest)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryChangesAndCursorAdvanceAreAtomic(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	changes := LocalTaskDiscovery{
-		DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-3", Body: []byte(`{"changes":[{"operation":"updated"},{"operation":"removed"}]}`),
-		Changes: []LocalTaskDiscoveryChange{
-			{Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")},
-			{Cursor: "cursor-3", Operation: "removed", TaskID: "task-b", TenantID: "tenant-a", TenantKey: "tenant-key-a"},
-		},
-	}
-	if err := st.ApplyLocalTaskDiscovery(changes); err != nil {
-		t.Fatal(err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || !exists || cursor != "cursor-3" {
-		t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err)
-	}
-	assignments, err := st.LocalTaskAssignments("d-1")
-	if err != nil || len(assignments) != 2 {
-		t.Fatalf("assignments=%+v err=%v", assignments, err)
-	}
-	if assignments[0].TaskRevision != 2 || assignments[0].Status != "paused" || assignments[0].AdmissionState != "paused" || !assignments[1].Removed || assignments[1].AdmissionState != "removed" {
-		t.Fatalf("change application = %+v", assignments)
-	}
-	var stored []byte
-	if err := st.db.QueryRow(`SELECT body FROM local_v02_task_discovery_response WHERE dispatcher_id=?`, "d-1").Scan(&stored); err != nil || string(stored) != string(changes.Body) {
-		t.Fatalf("stored response=%s err=%v", stored, err)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryNoChangeKeepsCursor(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	if err := st.ApplyLocalTaskDiscovery(LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-1", Body: []byte(`{"changes":[]}`)}); err != nil {
-		t.Fatal(err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || !exists || cursor != "cursor-1" {
-		t.Fatalf("empty change cursor=%q exists=%v err=%v", cursor, exists, err)
-	}
-}
-
-func TestLocalTaskDiscoveryFailureClosesAdmissionUntilFreshResponse(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	if err := st.CloseLocalTaskDiscoveryAdmission("d-1"); err != nil {
-		t.Fatal(err)
-	}
-	now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC)
-	bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now)
-	record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a")
-	record.DispatcherID = "d-1"
-	if _, err := st.PersistLocalCommand(record); !errors.Is(err, ErrLocalDiscoveryUnavailable) {
-		t.Fatalf("admission despite failed discovery: %v", err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || !exists || cursor != "cursor-1" {
-		t.Fatalf("lost durable cursor: %q exists=%v err=%v", cursor, exists, err)
-	}
-	if err := st.ApplyLocalTaskDiscovery(LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-1", Body: []byte(`{"changes":[]}`)}); err != nil {
-		t.Fatal(err)
-	}
-	if _, err := st.PersistLocalCommand(record); !errors.Is(err, ErrNoCapacity) {
-		t.Fatalf("fresh discovery should pass the discovery gate and reach the quota check, got %v", err)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryRejectsCursorMismatchWithoutAdvancing(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	bad := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "wrong-cursor", NextCursor: "cursor-2", Body: []byte(`{"changes":[{}]}`), Changes: []LocalTaskDiscoveryChange{{Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")}}}
-	if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrLocalDiscoveryCursorMismatch) {
-		t.Fatalf("cursor mismatch error=%v", err)
-	}
-	cursor, _, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || cursor != "cursor-1" {
-		t.Fatalf("cursor after rejected response=%q err=%v", cursor, err)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryRollsBackEarlierChangesOnLaterConflict(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	bad := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-3", Body: []byte(`{"changes":[{},{}]}`), Changes: []LocalTaskDiscoveryChange{
-		{Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")},
-		{Cursor: "cursor-3", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running")},
-	}}
-	if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrConfigRevisionRollback) {
-		t.Fatalf("revision rollback error=%v", err)
-	}
-	cursor, _, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || cursor != "cursor-1" {
-		t.Fatalf("cursor after rollback=%q err=%v", cursor, err)
-	}
-	task, err := st.LocalTaskAssignment("d-1", "task-a")
-	if err != nil || task.TaskRevision != 1 || task.Status != "running" || task.AdmissionState != "running" {
-		t.Fatalf("task after rollback=%+v err=%v", task, err)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryRejectsMoreThan256AssignedTasksWithoutAdvancing(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	tasks := make([]LocalDiscoveredTask, 256)
-	for i := range tasks {
-		tasks[i] = localDiscoveredTask("d-1", fmt.Sprintf("task-%03d", i), "tenant-a", "tenant-key-a", 1, "running")
-	}
-	if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[]}`), tasks...)); err != nil {
-		t.Fatal(err)
-	}
-	change := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-2", Body: []byte(`{"changes":[{}]}`), Changes: []LocalTaskDiscoveryChange{
-		{Cursor: "cursor-2", Operation: "assigned", Task: localDiscoveredTask("d-1", "task-256", "tenant-a", "tenant-key-a", 1, "running")},
-	}}
-	if err := st.ApplyLocalTaskDiscovery(change); !errors.Is(err, ErrLocalDiscoveryTaskLimit) {
-		t.Fatalf("task overflow error=%v", err)
-	}
-	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
-	if err != nil || !exists || cursor != "cursor-1" {
-		t.Fatalf("task overflow advanced cursor: %q exists=%v err=%v", cursor, exists, err)
-	}
-	if _, err := st.LocalTaskAssignment("d-1", "task-256"); !errors.Is(err, sql.ErrNoRows) {
-		t.Fatalf("task overflow persisted partial assignment: %v", err)
-	}
-}
-
-func TestApplyLocalTaskDiscoverySnapshotMarksOmittedTasksRemoved(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	seedLocalDiscovery(t, st)
-	if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-2", []byte(`{"tasks":[{}]}`),
-		localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"),
-	)); err != nil {
-		t.Fatal(err)
-	}
-	task, err := st.LocalTaskAssignment("d-1", "task-b")
-	if err != nil || !task.Removed || task.AdmissionState != "removed" {
-		t.Fatalf("omitted task=%+v err=%v", task, err)
-	}
-}
-
-func TestApplyLocalTaskDiscoveryRejectsTenantBindingConflict(t *testing.T) {
-	st := openLocalDiscoveryTestStore(t)
-	bad := localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{},{}]}`),
-		localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"),
-		localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-b", 1, "running"),
-	)
-	if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrTenantBindingConflict) {
-		t.Fatalf("tenant binding error=%v", err)
-	}
-	if _, exists, err := st.LocalTaskDiscoveryCursor("d-1"); err != nil || exists {
-		t.Fatalf("partial snapshot cursor persisted=%v err=%v", exists, err)
-	}
-}
-
-func TestPersistLocalCommandRequiresRunningDiscoveredTask(t *testing.T) {
-	for _, tc := range []struct {
-		name       string
-		discovered bool
-		status     string
-		wantErr    error
-	}{
-		{name: "paused", discovered: true, status: "paused", wantErr: ErrLocalTaskPaused},
-		{name: "stopped", discovered: true, status: "stopped", wantErr: ErrLocalTaskStopped},
-		{name: "no-discovery", wantErr: ErrLocalDiscoveryUnavailable},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			st := openLocalDiscoveryTestStore(t)
-			if tc.discovered {
-				if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{}]}`),
-					localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, tc.status),
-				)); err != nil {
-					t.Fatal(err)
-				}
-			}
-			now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC)
-			bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now)
-			record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a")
-			record.DispatcherID = "d-1"
-			if _, err := st.PersistLocalCommand(record); !errors.Is(err, tc.wantErr) {
-				t.Fatalf("PersistLocalCommand error=%v, want %v", err, tc.wantErr)
-			}
-			for _, table := range []string{"inbox", "outbox", "tasks", "reservations"} {
-				var count int
-				if err := st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != 0 {
-					t.Fatalf("table %s rows=%d err=%v after blocked admission", table, count, err)
-				}
-			}
-		})
-	}
-}
-
-func openLocalDiscoveryTestStore(t *testing.T) *Store {
-	t.Helper()
-	st, err := Open(filepath.Join(t.TempDir(), "dispatcher.db"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(func() { _ = st.Close() })
-	return st
-}
-
-func seedLocalCommandAssignments(t *testing.T, st *Store, records ...LocalCommandRecord) {
-	t.Helper()
-	if len(records) == 0 {
-		return
-	}
-	dispatcherID := records[0].DispatcherID
-	tasks := make([]LocalDiscoveredTask, 0, len(records))
-	seen := make(map[string]struct{}, len(records))
-	for _, record := range records {
-		if record.DispatcherID != dispatcherID {
-			t.Fatalf("test commands use different Dispatcher IDs: %q and %q", dispatcherID, record.DispatcherID)
-		}
-		if _, exists := seen[record.TaskID]; exists {
-			continue
-		}
-		seen[record.TaskID] = struct{}{}
-		revision := int64(1)
-		if record.Admission != nil && record.Admission.Task.TaskRevision > 0 {
-			revision = record.Admission.Task.TaskRevision
-		}
-		tasks = append(tasks, localDiscoveredTask(dispatcherID, record.TaskID, record.TenantID, record.TenantKey, revision, "running"))
-	}
-	if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow(dispatcherID, "test-cursor", []byte(`{"tasks":[]}`), tasks...)); err != nil {
-		t.Fatal(err)
-	}
-}
-
-func seedLocalDiscovery(t *testing.T, st *Store) {
-	t.Helper()
-	if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{},{}]}`),
-		localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"),
-		localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 1, "running"),
-	)); err != nil {
-		t.Fatal(err)
-	}
-}
-
-func localSnapshotWindow(dispatcherID, cursor string, body []byte, tasks ...LocalDiscoveredTask) LocalTaskDiscovery {
-	return LocalTaskDiscovery{DispatcherID: dispatcherID, Mode: "snapshot", Cursor: cursor, Body: body, Tasks: tasks}
-}
-
-func localDiscoveredTask(_ string, taskID, tenantID, tenantKey string, revision int64, status string) LocalDiscoveredTask {
-	return LocalDiscoveredTask{TaskID: taskID, TenantID: tenantID, TenantKey: tenantKey, TaskRevision: revision, Status: status}
-}

internal/store/local_v03_admission_test.go

@@ -0,0 +1,40 @@
+package store
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestEventDiscoveryAdmissionRequiresRunningTask(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		discovered bool
+		status     string
+		wantErr    error
+	}{
+		{name: "paused", discovered: true, status: "paused", wantErr: ErrLocalTaskPaused},
+		{name: "stopped", discovered: true, status: "stopped", wantErr: ErrLocalTaskStopped},
+		{name: "removed", discovered: true, status: "removed", wantErr: ErrLocalTaskStopped},
+		{name: "no discovery", wantErr: ErrLocalDiscoveryUnavailable},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			st := openLocalDiscoveryTestStore(t)
+			if tc.discovered {
+				seedLocalEventTasks(t, st, "d-1", eventTask("task-a", tc.status, 1))
+			}
+			now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC)
+			bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now)
+			record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a")
+			if _, err := st.PersistLocalCommand(record); !errors.Is(err, tc.wantErr) {
+				t.Fatalf("PersistLocalCommand error=%v, want %v", err, tc.wantErr)
+			}
+			for _, table := range []string{"inbox", "outbox", "tasks", "reservations"} {
+				var count int
+				if err := st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != 0 {
+					t.Fatalf("table %s rows=%d err=%v after blocked admission", table, count, err)
+				}
+			}
+		})
+	}
+}

internal/store/local_v03_config_gate_test.go

@@ -0,0 +1,35 @@
+package store
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestEventConfigReadRequiresCompleteDiscoveryAndTenantBinding(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a"); !errors.Is(err, ErrLocalDiscoveryUnavailable) {
+		t.Fatalf("partial discovery admitted config read: %v", err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "1")); err != nil {
+		t.Fatal(err)
+	}
+	assignment, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a")
+	if err != nil || assignment.Status != "running" || assignment.TenantKey != "tenant-key-a" {
+		t.Fatalf("assignment=%+v err=%v", assignment, err)
+	}
+	if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "another-tenant"); !errors.Is(err, ErrTenantBindingConflict) {
+		t.Fatalf("cross-tenant config read: %v", err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "removed", 1))); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("2", "2")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a"); !errors.Is(err, ErrLocalTaskRemoved) {
+		t.Fatalf("removed task config read: %v", err)
+	}
+}

internal/store/local_v03_discovery.go

@@ -0,0 +1,165 @@
+package store
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+)
+
+const localEventPageLimit = 256
+
+// LocalTaskDiscoveryPage is one complete HTTP response, not an entire task list.
+// SaaS guarantees response ordering/completeness; its response-level cursor is
+// the only event position supplied by the project-local v0.3 proposal.
+type LocalTaskDiscoveryPage struct {
+	DispatcherID string
+	FromCursor   string
+	NextCursor   string
+	Tasks        []LocalDiscoveredTask
+	Body         json.RawMessage
+	ObservedAt   time.Time
+}
+
+func eventCursorValue(cursor string) (uint64, error) {
+	value, err := strconv.ParseUint(cursor, 10, 64)
+	if err != nil || strconv.FormatUint(value, 10) != cursor {
+		return 0, errors.New("event cursor must be a canonical decimal uint64")
+	}
+	return value, nil
+}
+
+// ApplyLocalTaskDiscoveryPage commits assignments, the exact page evidence and
+// its cursor together. A nonempty page never opens new execution admission;
+// only an empty final page marks the initial catch-up/current poll complete.
+func (s *Store) ApplyLocalTaskDiscoveryPage(page LocalTaskDiscoveryPage) error {
+	from, err := eventCursorValue(page.FromCursor)
+	if err != nil {
+		return fmt.Errorf("invalid task-discovery after cursor: %w", err)
+	}
+	to, err := eventCursorValue(page.NextCursor)
+	if err != nil {
+		return fmt.Errorf("invalid task-discovery next cursor: %w", err)
+	}
+	if page.DispatcherID == "" || page.ObservedAt.IsZero() || len(page.Tasks) > localEventPageLimit || !json.Valid(page.Body) ||
+		(len(page.Tasks) == 0 && to != from) || (len(page.Tasks) != 0 && to <= from) {
+		return ErrLocalDiscoveryUnavailable
+	}
+	seen := make(map[string]struct{}, len(page.Tasks))
+	for _, task := range page.Tasks {
+		if err := validateLocalDiscoveredTask(page.DispatcherID, task); err != nil {
+			return err
+		}
+		if _, duplicate := seen[task.TaskID]; duplicate {
+			return fmt.Errorf("task-discovery page has duplicate task ID %q", task.TaskID)
+		}
+		seen[task.TaskID] = struct{}{}
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	var legacy int
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, page.DispatcherID).Scan(&legacy); err != nil {
+		return err
+	}
+	if legacy != 0 {
+		return fmt.Errorf("%w: v0.2 discovery checkpoint must be drained before switching", ErrLocalDiscoveryUnavailable)
+	}
+	var cursor string
+	err = tx.QueryRow(`SELECT cursor FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, page.DispatcherID).Scan(&cursor)
+	if errors.Is(err, sql.ErrNoRows) {
+		var existing int
+		if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=?`, page.DispatcherID).Scan(&existing); err != nil {
+			return err
+		}
+		if existing != 0 {
+			return fmt.Errorf("%w: existing assignments have no v0.3 checkpoint", ErrLocalDiscoveryUnavailable)
+		}
+		if page.FromCursor != "0" {
+			return ErrLocalDiscoveryCursorMismatch
+		}
+	} else if err != nil {
+		return err
+	} else if cursor != page.FromCursor {
+		return ErrLocalDiscoveryCursorMismatch
+	}
+	now := page.ObservedAt.UTC().Format(time.RFC3339Nano)
+	for _, task := range page.Tasks {
+		if err := bindLocalTenant(tx, page.DispatcherID, task.TenantID, task.TenantKey, page.ObservedAt); err != nil {
+			return err
+		}
+		if err := applyLocalDiscoveredTask(tx, page.DispatcherID, task, now); err != nil {
+			return err
+		}
+	}
+	var active int
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, page.DispatcherID).Scan(&active); err != nil {
+		return err
+	}
+	if active > localEventPageLimit {
+		return ErrLocalDiscoveryTaskLimit
+	}
+	ready := 0
+	if len(page.Tasks) == 0 {
+		ready = 1
+	}
+	sum := sha256.Sum256(page.Body)
+	_, err = tx.Exec(`INSERT INTO local_v03_task_discovery_state(dispatcher_id,cursor,ready,body,content_sha256,updated_at)
+		VALUES(?,?,?,?,?,?) ON CONFLICT(dispatcher_id) DO UPDATE SET cursor=excluded.cursor,ready=excluded.ready,
+		body=excluded.body,content_sha256=excluded.content_sha256,updated_at=excluded.updated_at`,
+		page.DispatcherID, page.NextCursor, ready, []byte(page.Body), hex.EncodeToString(sum[:]), now)
+	if err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+// LocalDiscoveredTaskForConfig binds a config read to a fully caught-up,
+// durable task assignment. A partial page cannot authorize configuration.
+func (s *Store) LocalDiscoveredTaskForConfig(dispatcherID, taskID, tenantID string) (LocalTaskAssignment, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var ready int
+	var observedAt string
+	err := s.db.QueryRow(`SELECT ready,updated_at FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready, &observedAt)
+	if errors.Is(err, sql.ErrNoRows) || (err == nil && ready != 1) {
+		return LocalTaskAssignment{}, ErrLocalDiscoveryUnavailable
+	}
+	if err != nil {
+		return LocalTaskAssignment{}, err
+	}
+	var assignment LocalTaskAssignment
+	var removed int
+	err = s.db.QueryRow(`SELECT dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed,
+		queue_exchange,routing_key,binding_key,queue_name,updated_at FROM local_v01_task_assignments WHERE dispatcher_id=? AND task_id=?`,
+		dispatcherID, taskID).Scan(&assignment.DispatcherID, &assignment.TaskID, &assignment.TenantID, &assignment.TenantKey,
+		&assignment.TaskRevision, &assignment.Status, &assignment.AdmissionState, &removed, &assignment.Queue.Exchange,
+		&assignment.Queue.RoutingKey, &assignment.Queue.BindingKey, &assignment.Queue.QueueName, &assignment.UpdatedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return LocalTaskAssignment{}, ErrLocalDiscoveryTaskMissing
+	}
+	if err != nil {
+		return LocalTaskAssignment{}, err
+	}
+	if assignment.TenantID != tenantID {
+		return LocalTaskAssignment{}, ErrTenantBindingConflict
+	}
+	assignment.Removed = removed != 0
+	if assignment.Removed {
+		return LocalTaskAssignment{}, ErrLocalTaskRemoved
+	}
+	assignment.DiscoveryObservedAt, err = time.Parse(time.RFC3339Nano, observedAt)
+	if err != nil {
+		return LocalTaskAssignment{}, fmt.Errorf("invalid durable discovery observation time: %w", err)
+	}
+	return assignment, nil
+}

internal/store/local_v03_discovery_test.go

@@ -0,0 +1,92 @@
+package store
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"testing"
+	"time"
+)
+
+func eventPage(after, next string, tasks ...LocalDiscoveredTask) LocalTaskDiscoveryPage {
+	return LocalTaskDiscoveryPage{
+		DispatcherID: "d-1", FromCursor: after, NextCursor: next,
+		Tasks: tasks, Body: []byte(`{"schema_version":"task-discovery.v0.3-proposal","tasks":[]}`),
+		ObservedAt: time.Now().UTC(),
+	}
+}
+
+func eventTask(id, status string, revision int64) LocalDiscoveredTask {
+	return LocalDiscoveredTask{TaskID: id, TenantID: "tenant-a", TenantKey: "tenant-key-a", Status: status, TaskRevision: revision}
+}
+
+func TestEventPagesPersistCursorAssignmentAndBootGateTogether(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	first := eventPage("0", "1", eventTask("task-a", "running", 1))
+	if err := st.ApplyLocalTaskDiscoveryPage(first); err != nil {
+		t.Fatal(err)
+	}
+	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
+	if err != nil || !exists || cursor != "1" {
+		t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err)
+	}
+	var ready int
+	var saved []byte
+	var digest string
+	if err := st.db.QueryRow(`SELECT ready,body,content_sha256 FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, "d-1").Scan(&ready, &saved, &digest); err != nil {
+		t.Fatal(err)
+	}
+	sum := sha256.Sum256(first.Body)
+	if ready != 0 || string(saved) != string(first.Body) || digest != hex.EncodeToString(sum[:]) {
+		t.Fatalf("partial page admitted or changed evidence: ready=%d body=%s hash=%s", ready, saved, digest)
+	}
+	assignments, err := st.LocalTaskAssignments("d-1")
+	if err != nil || len(assignments) != 1 || assignments[0].Status != "running" || assignments[0].Queue.QueueName != "agent-call.d.d-1.task.task-a.v3" {
+		t.Fatalf("assignment=%+v err=%v", assignments, err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "1")); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.db.QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, "d-1").Scan(&ready); err != nil || ready != 1 {
+		t.Fatalf("completed page ready=%d err=%v", ready, err)
+	}
+}
+
+func TestEventPageAllowsSameTaskNewStatusAndRemovedTombstone(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	for _, page := range []LocalTaskDiscoveryPage{
+		eventPage("0", "1", eventTask("task-a", "running", 1)),
+		eventPage("1", "2", eventTask("task-a", "paused", 1)),
+		eventPage("2", "3", eventTask("task-a", "removed", 1)),
+		eventPage("3", "3"),
+	} {
+		if err := st.ApplyLocalTaskDiscoveryPage(page); err != nil {
+			t.Fatal(err)
+		}
+	}
+	assignments, err := st.LocalTaskAssignments("d-1")
+	if err != nil || len(assignments) != 1 || !assignments[0].Removed || assignments[0].Status != "removed" || assignments[0].AdmissionState != "removed" {
+		t.Fatalf("removed task=%+v err=%v", assignments, err)
+	}
+	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
+	if err != nil || !exists || cursor != "3" {
+		t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "4", eventTask("task-a", "running", 1))); !errors.Is(err, ErrLocalDiscoveryCursorMismatch) {
+		t.Fatalf("old cursor unexpectedly accepted: %v", err)
+	}
+}
+
+func TestEventPageCannotInterpretOldSnapshotCursor(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	if _, err := st.db.Exec(`INSERT INTO local_v02_task_discovery_state(dispatcher_id,cursor,last_mode,ready,updated_at) VALUES('d-1','opaque-old','snapshot',1,'now')`); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); !errors.Is(err, ErrLocalDiscoveryUnavailable) {
+		t.Fatalf("legacy cursor should block, got %v", err)
+	}
+	var count int
+	if err := st.db.QueryRow(`SELECT COUNT(*) FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("legacy cursor created new checkpoint: count=%d err=%v", count, err)
+	}
+}

internal/store/local_v03_failure_test.go

@@ -0,0 +1,78 @@
+package store
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestEventPageWriteFailureRollsBackAssignmentCursorAndEvidence(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil {
+		t.Fatal(err)
+	}
+	var before []byte
+	if err := st.db.QueryRow(`SELECT body FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&before); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.db.Exec(`CREATE TRIGGER reject_discovery_update BEFORE UPDATE ON local_v01_task_assignments
+		BEGIN SELECT RAISE(ABORT, 'forced assignment write failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "paused", 2)))
+	if err == nil || !strings.Contains(err.Error(), "forced assignment write failure") {
+		t.Fatalf("assignment write failure was swallowed: %v", err)
+	}
+	cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1")
+	if err != nil || !exists || cursor != "1" {
+		t.Fatalf("failed page advanced cursor: %q exists=%v err=%v", cursor, exists, err)
+	}
+	assignment, err := st.LocalTaskAssignment("d-1", "task-a")
+	if err != nil || assignment.Status != "running" || assignment.TaskRevision != 1 {
+		t.Fatalf("failed page changed assignment: %+v err=%v", assignment, err)
+	}
+	var after []byte
+	if err := st.db.QueryRow(`SELECT body FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&after); err != nil || string(after) != string(before) {
+		t.Fatalf("failed page replaced saved evidence: before=%s after=%s err=%v", before, after, err)
+	}
+}
+
+func TestEventPageCannotUndoMQPauseStopOrRemoval(t *testing.T) {
+	st := openLocalDiscoveryTestStore(t)
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.SetLocalTaskAdmissionBarrier("d-1", "task-a", "tenant-a", "tenant-key-a", "paused"); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "running", 2))); err != nil {
+		t.Fatal(err)
+	}
+	assignment, err := st.LocalTaskAssignment("d-1", "task-a")
+	if err != nil || assignment.AdmissionState != "paused" {
+		t.Fatalf("stale running discovery overrode MQ pause: %+v err=%v", assignment, err)
+	}
+	if _, err := st.SetLocalTaskAdmissionBarrier("d-1", "task-a", "tenant-a", "tenant-key-a", "stopped"); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("2", "3", eventTask("task-a", "running", 3))); err != nil {
+		t.Fatal(err)
+	}
+	assignment, err = st.LocalTaskAssignment("d-1", "task-a")
+	if err != nil || assignment.AdmissionState != "stopped" {
+		t.Fatalf("stale running discovery overrode MQ stop: %+v err=%v", assignment, err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("3", "4", eventTask("task-a", "removed", 3))); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("4", "5", eventTask("task-a", "running", 4))); !errors.Is(err, ErrLocalTaskRemoved) {
+		t.Fatalf("removed task was revived: %v", err)
+	}
+	if err := st.ApplyLocalTaskDiscoveryPage(eventPage("4", "4")); err != nil {
+		t.Fatal(err)
+	}
+	assignments, err := st.LocalTaskAssignments("d-1")
+	if err != nil || len(assignments) != 1 || !assignments[0].Removed || assignments[0].AdmissionState != "removed" {
+		t.Fatalf("later event revived removed task: %+v err=%v", assignments, err)
+	}
+}

internal/store/migrations/018_local_v03_task_discovery.sql

@@ -0,0 +1,10 @@
+-- Separate event-cursor checkpoint: never reinterpret a v0.2 snapshot/change cursor.
+-- The older tables remain untouched for controlled drain/diagnosis, not fallback.
+CREATE TABLE IF NOT EXISTS local_v03_task_discovery_state (
+    dispatcher_id TEXT PRIMARY KEY,
+    cursor TEXT NOT NULL,
+    ready INTEGER NOT NULL CHECK (ready IN (0, 1)),
+    body BLOB NOT NULL,
+    content_sha256 TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);