test(runtime-b): prove provider-side cancellation

runtime-b/README.md

@@ -84,8 +84,15 @@ Success means every package prints `ok`, the race detector and vet exit with
 code `0`, and the image builds. Required tests prove official A2A client
 interoperability, strict one-line SSE frames, deterministic message and stream
 results, task history bounds, same-task cancellation, concurrent identity
-isolation, Router-only nested lineage, and a readiness request that creates no
-task state.
+isolation, Router-only nested lineage, Provider-side cancellation observation,
+and a readiness request that creates no task state. The deterministic
+`cancel-observed` fixture consumes a non-sensitive marker and returns only a
+boolean and Provider-observed request count, including rejected duplicate
+attempts. Task access is scoped by the authenticated Workspace, exact Release
+provenance, capability, and Invocation; observations use the same scope without
+the Invocation so a later managed call can read them. The fixture exists so
+Stack can prove exactly one Router `tasks/cancel` reached this Provider without
+exposing a direct Provider inspection path.
 
 ## Run Runtime B
 

runtime-b/fixture.go

@@ -11,13 +11,14 @@ import (
 type fixtureKind string
 
 const (
-	fixtureSuccess       fixtureKind = "success"
-	fixtureStreamSuccess fixtureKind = "stream-success"
-	fixtureFailure       fixtureKind = "failure"
-	fixtureProtocol      fixtureKind = "protocol"
-	fixtureHold          fixtureKind = "hold"
-	fixtureInterrupted   fixtureKind = "interrupted"
-	fixtureNested        fixtureKind = "nested"
+	fixtureSuccess        fixtureKind = "success"
+	fixtureStreamSuccess  fixtureKind = "stream-success"
+	fixtureFailure        fixtureKind = "failure"
+	fixtureProtocol       fixtureKind = "protocol"
+	fixtureHold           fixtureKind = "hold"
+	fixtureCancelObserved fixtureKind = "cancel-observed"
+	fixtureInterrupted    fixtureKind = "interrupted"
+	fixtureNested         fixtureKind = "nested"
 )
 
 var errFixtureFailure = errors.New("runtime-b deterministic fixture failure")
@@ -63,7 +64,7 @@ func parseFixture(params *a2a.MessageSendParams) (fixtureRequest, error) {
 
 	kind := fixtureKind(fixture)
 	switch kind {
-	case fixtureSuccess, fixtureStreamSuccess, fixtureFailure, fixtureProtocol, fixtureHold, fixtureInterrupted, fixtureNested:
+	case fixtureSuccess, fixtureStreamSuccess, fixtureFailure, fixtureProtocol, fixtureHold, fixtureCancelObserved, fixtureInterrupted, fixtureNested:
 	default:
 		return fixtureRequest{}, invalidParams("fixture is not supported")
 	}

runtime-b/handler.go

@@ -7,28 +7,32 @@ import (
 	"sync"
 
 	agentsdk "github.com/NeKiro-project/nekiro-sdk-go/agent"
+	"github.com/NeKiro-project/nekiro-sdk-go/agent/routerauth"
 	"github.com/a2aproject/a2a-go/a2a"
 	"github.com/a2aproject/a2a-go/a2asrv"
 )
 
 type runtimeTask struct {
-	task   *a2a.Task
-	cancel chan struct{}
+	task              *a2a.Task
+	cancel            chan struct{}
+	owner             string
+	cancelObservation string
 }
 
 // Handler implements the active A2A Profile for the deterministic Runtime B sample.
 type Handler struct {
-	mu         sync.RWMutex
-	tasks      map[a2a.TaskID]*runtimeTask
-	agentID    string
-	instanceID string
-	nested     *nestedService
+	mu            sync.RWMutex
+	tasks         map[a2a.TaskID]*runtimeTask
+	cancellations map[string]int
+	agentID       string
+	instanceID    string
+	nested        *nestedService
 }
 
 var _ a2asrv.RequestHandler = (*Handler)(nil)
 
 func NewHandler() *Handler {
-	return &Handler{tasks: make(map[a2a.TaskID]*runtimeTask), instanceID: "runtime-b"}
+	return &Handler{tasks: make(map[a2a.TaskID]*runtimeTask), cancellations: make(map[string]int), instanceID: "runtime-b"}
 }
 
 // NewConfiguredHandler creates the production Runtime B handler with one
@@ -45,7 +49,7 @@ func NewConfiguredHandler(config Config, doer agentsdk.HTTPDoer) (*Handler, erro
 	if err != nil {
 		return nil, err
 	}
-	return &Handler{tasks: make(map[a2a.TaskID]*runtimeTask), agentID: config.AgentID, instanceID: config.InstanceID, nested: nested}, nil
+	return &Handler{tasks: make(map[a2a.TaskID]*runtimeTask), cancellations: make(map[string]int), agentID: config.AgentID, instanceID: config.InstanceID, nested: nested}, nil
 }
 
 func (h *Handler) OnSendMessage(ctx context.Context, params *a2a.MessageSendParams) (a2a.SendMessageResult, error) {
@@ -56,6 +60,12 @@ func (h *Handler) OnSendMessage(ctx context.Context, params *a2a.MessageSendPara
 	switch request.kind {
 	case fixtureSuccess:
 		return h.successMessage(params.Message, request), nil
+	case fixtureCancelObserved:
+		marker, ok := request.value.(string)
+		if !ok || marker == "" {
+			return nil, invalidParams("cancel observation marker must be a non-empty string")
+		}
+		return h.cancelObservationMessage(params.Message, h.takeCancellationObservation(cancellationObservationKey(ctx, marker))), nil
 	case fixtureNested:
 		if h.nested == nil {
 			return nil, invalidParams("nested fixture is not configured")
@@ -104,14 +114,16 @@ func (h *Handler) OnSendMessageStream(ctx context.Context, params *a2a.MessageSe
 			return
 		}
 
-		task, err := h.createWorkingTask(params.Message)
+		task, err := h.createWorkingTask(ctx, params.Message, request)
 		if err != nil {
 			yield(nil, err)
 			return
 		}
 		terminal := false
 		defer func() {
-			if !terminal {
+			// A hold task remains addressable after its stream disconnects so the
+			// Router's single bounded tasks/cancel call can reach the same task.
+			if !terminal && request.kind != fixtureHold {
 				h.removeWorkingTask(task.task.ID)
 			}
 		}()
@@ -175,7 +187,7 @@ func (h *Handler) OnSendMessageStream(ctx context.Context, params *a2a.MessageSe
 	}
 }
 
-func (h *Handler) OnGetTask(_ context.Context, query *a2a.TaskQueryParams) (*a2a.Task, error) {
+func (h *Handler) OnGetTask(ctx context.Context, query *a2a.TaskQueryParams) (*a2a.Task, error) {
 	if query == nil || query.ID == "" {
 		return nil, invalidParams("task id is required")
 	}
@@ -185,7 +197,7 @@ func (h *Handler) OnGetTask(_ context.Context, query *a2a.TaskQueryParams) (*a2a
 
 	h.mu.RLock()
 	stored, exists := h.tasks[query.ID]
-	if !exists {
+	if !exists || stored.owner != taskOwner(ctx) {
 		h.mu.RUnlock()
 		return nil, a2a.ErrTaskNotFound
 	}
@@ -200,16 +212,22 @@ func (h *Handler) OnGetTask(_ context.Context, query *a2a.TaskQueryParams) (*a2a
 	return task, nil
 }
 
-func (h *Handler) OnCancelTask(_ context.Context, params *a2a.TaskIDParams) (*a2a.Task, error) {
+func (h *Handler) OnCancelTask(ctx context.Context, params *a2a.TaskIDParams) (*a2a.Task, error) {
 	if params == nil || params.ID == "" {
 		return nil, invalidParams("task id is required")
 	}
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	stored, exists := h.tasks[params.ID]
-	if !exists {
+	if !exists || stored.owner != taskOwner(ctx) {
 		return nil, a2a.ErrTaskNotFound
 	}
+	if stored.cancelObservation != "" {
+		if h.cancellations == nil {
+			h.cancellations = make(map[string]int)
+		}
+		h.cancellations[stored.cancelObservation]++
+	}
 	if stored.task.Status.State != a2a.TaskStateWorking {
 		return nil, a2a.ErrTaskNotCancelable
 	}
@@ -218,7 +236,7 @@ func (h *Handler) OnCancelTask(_ context.Context, params *a2a.TaskIDParams) (*a2
 	return cloneTask(stored.task), nil
 }
 
-func (h *Handler) createWorkingTask(message *a2a.Message) (*runtimeTask, error) {
+func (h *Handler) createWorkingTask(ctx context.Context, message *a2a.Message, request fixtureRequest) (*runtimeTask, error) {
 	taskID := a2a.TaskID(derivedID("task", message.ID))
 	contextID := message.ContextID
 	if contextID == "" {
@@ -231,7 +249,12 @@ func (h *Handler) createWorkingTask(message *a2a.Message) (*runtimeTask, error)
 			Status:    a2a.TaskStatus{State: a2a.TaskStateWorking},
 			History:   []*a2a.Message{cloneMessage(message)},
 		},
-		cancel: make(chan struct{}),
+		cancel: make(chan struct{}), owner: taskOwner(ctx),
+	}
+	if request.kind == fixtureHold {
+		if marker, ok := request.value.(string); ok && marker != "" {
+			stored.cancelObservation = cancellationObservationKey(ctx, marker)
+		}
 	}
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -242,6 +265,40 @@ func (h *Handler) createWorkingTask(message *a2a.Message) (*runtimeTask, error)
 	return stored, nil
 }
 
+func cancellationObservationKey(ctx context.Context, marker string) string {
+	owner := observationOwner(ctx)
+	if owner == "" {
+		return marker
+	}
+	return owner + "\x00" + marker
+}
+
+func taskOwner(ctx context.Context) string {
+	owner := observationOwner(ctx)
+	if owner == "" {
+		return ""
+	}
+	claims, _ := routerauth.ClaimsFromContext(ctx)
+	return owner + "\x00" + claims.InvocationID
+}
+
+func observationOwner(ctx context.Context) string {
+	claims, ok := routerauth.ClaimsFromContext(ctx)
+	if !ok {
+		return ""
+	}
+	return claims.WorkspaceID + "\x00" + claims.AgentID + "\x00" + claims.AgentVersion + "\x00" +
+		claims.ReleaseID + "\x00" + claims.CardDigest + "\x00" + claims.Capability
+}
+
+func (h *Handler) takeCancellationObservation(marker string) int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	count := h.cancellations[marker]
+	delete(h.cancellations, marker)
+	return count
+}
+
 func (h *Handler) completeTask(taskID a2a.TaskID) (a2a.TaskState, error) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -295,6 +352,20 @@ func (h *Handler) successMessage(input *a2a.Message, request fixtureRequest) *a2
 	}
 }
 
+func (h *Handler) cancelObservationMessage(input *a2a.Message, count int) *a2a.Message {
+	contextID := input.ContextID
+	if contextID == "" {
+		contextID = derivedID("context", input.ID)
+	}
+	return &a2a.Message{
+		ID: derivedID("message", input.ID), ContextID: contextID, Role: a2a.MessageRoleAgent,
+		Parts: []a2a.Part{a2a.DataPart{Data: map[string]any{
+			"agent": "runtime-b", "instanceId": h.instanceID, "fixture": string(fixtureCancelObserved),
+			"canceled": count > 0, "cancelCount": count,
+		}}},
+	}
+}
+
 func (h *Handler) streamMessage(task *a2a.Task, request fixtureRequest) *a2a.Message {
 	return &a2a.Message{
 		ID:        derivedID("stream-message", string(task.ID)),

runtime-b/handler_test.go

@@ -189,6 +189,63 @@ func TestHandlerHoldStreamCancelsSameTask(t *testing.T) {
 	}
 }
 
+func TestHandlerHoldTaskSurvivesDisconnectForOneObservedCancel(t *testing.T) {
+	handler := NewHandler()
+	ctx, disconnect := context.WithCancel(t.Context())
+	events := make(chan a2a.Event, 1)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for event, err := range handler.OnSendMessageStream(ctx, fixtureParams("hold-observed", fixtureHold, "marker-a")) {
+			if err != nil {
+				t.Errorf("hold stream: %v", err)
+				return
+			}
+			events <- event
+		}
+	}()
+	task := requireTaskEvent(t, receiveEvent(t, events))
+	disconnect()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("hold stream did not stop after disconnect")
+	}
+
+	canceled, err := handler.OnCancelTask(t.Context(), &a2a.TaskIDParams{ID: task.ID})
+	if err != nil || canceled.Status.State != a2a.TaskStateCanceled {
+		t.Fatalf("cancel disconnected task = (%#v, %v)", canceled, err)
+	}
+	assertCancelObservation(t, handler, "marker-a", true, 1)
+	assertCancelObservation(t, handler, "marker-a", false, 0)
+	if _, err := handler.OnCancelTask(t.Context(), &a2a.TaskIDParams{ID: task.ID}); !errors.Is(err, a2a.ErrTaskNotCancelable) {
+		t.Fatalf("second cancel = %v", err)
+	}
+	assertCancelObservation(t, handler, "marker-a", true, 1)
+}
+
+func TestHandlerCancelObservationRejectsInvalidMarker(t *testing.T) {
+	result, err := NewHandler().OnSendMessage(t.Context(), fixtureParams("cancel-observation", fixtureCancelObserved, map[string]any{"marker": "a"}))
+	if result != nil || !errors.Is(err, a2a.ErrInvalidParams) {
+		t.Fatalf("invalid marker result = (%#v, %v)", result, err)
+	}
+}
+
+func assertCancelObservation(t *testing.T, handler *Handler, marker string, canceled bool, count int) {
+	t.Helper()
+	result, err := handler.OnSendMessage(t.Context(), fixtureParams("observe-"+marker, fixtureCancelObserved, marker))
+	if err != nil {
+		t.Fatal(err)
+	}
+	part := requireDataPart(t, requireMessage(t, result).Parts[0])
+	if part.Data["fixture"] != string(fixtureCancelObserved) || part.Data["canceled"] != canceled || part.Data["cancelCount"] != count {
+		t.Fatalf("cancel observation = %#v", part.Data)
+	}
+	if _, leaked := part.Data["value"]; leaked {
+		t.Fatalf("cancel observation leaked marker: %#v", part.Data)
+	}
+}
+
 func TestHandlerStreamSuccessStopsContentWhenCancelWins(t *testing.T) {
 	for _, cancelAfter := range []int{1, 2, 3, 4} {
 		t.Run(fmt.Sprintf("after-event-%d", cancelAfter), func(t *testing.T) {
@@ -240,8 +297,12 @@ func TestHandlerHoldStreamContextTerminationDoesNotCreateTerminal(t *testing.T)
 		t.Fatal("context-terminated stream did not stop")
 	}
 	historyLength := 1
-	if _, err := handler.OnGetTask(t.Context(), &a2a.TaskQueryParams{ID: working.ID, HistoryLength: &historyLength}); !errors.Is(err, a2a.ErrTaskNotFound) {
-		t.Fatalf("context-terminated task = %v, want task not found", err)
+	stored, err := handler.OnGetTask(t.Context(), &a2a.TaskQueryParams{ID: working.ID, HistoryLength: &historyLength})
+	if err != nil || stored.Status.State != a2a.TaskStateWorking {
+		t.Fatalf("context-terminated task = (%#v, %v), want working task", stored, err)
+	}
+	if _, err := handler.OnCancelTask(t.Context(), &a2a.TaskIDParams{ID: working.ID}); err != nil {
+		t.Fatalf("cancel context-terminated task: %v", err)
 	}
 }