Unify cancellation confirmation budget

apps/daemon/internal/dispatch/cancellation_budget_test.go

@@ -0,0 +1,81 @@
+package dispatch_test
+
+import (
+	"context"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
+	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/dispatch"
+	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
+	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto/prototest"
+)
+
+type hangingCancelExecutor struct {
+	turn        *hangingCancelTurn
+	closeBudget time.Duration
+}
+
+func (e *hangingCancelExecutor) StartTurn(_ context.Context, id string, _ proto.MessageInput, out chan<- proto.Envelope) (agent.Turn, error) {
+	e.turn = &hangingCancelTurn{reusableTurn: &reusableTurn{id: id, out: out, settled: make(chan struct{})}}
+	return e.turn, nil
+}
+func (e *hangingCancelExecutor) Close(ctx context.Context) error {
+	deadline, _ := ctx.Deadline()
+	e.closeBudget = time.Until(deadline)
+	e.turn.finish()
+	return nil
+}
+
+type hangingCancelTurn struct {
+	*reusableTurn
+	cancelBudget time.Duration
+}
+
+func (turn *hangingCancelTurn) Cancel(ctx context.Context) error {
+	turn.cancels.Add(1)
+	deadline, _ := ctx.Deadline()
+	turn.cancelBudget = time.Until(deadline)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func TestNativeCancellationTimeoutCannotBecomeAppliedAfterClose(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		owner := &hangingCancelExecutor{}
+		reg := agent.NewRegistry()
+		registerExecutorKind(reg, proto.SupportedAgentKind{Kind: "reusable", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{EnvironmentNone: proto.CapabilitySupported})}, func(context.Context, agent.PrepareRequest) (agent.Executor, error) { return owner, nil })
+		sender := &recSender{}
+		r, err := dispatch.New(dispatch.Config{Registry: reg, Sender: sender, IdleTimeout: time.Hour})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer r.Shutdown(t.Context())
+		assign(t, r, preparationSessionID, "")
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "request", executorRequest())); err != nil {
+			t.Fatal(err)
+		}
+		ready := waitPreparationStatus(t, sender, "request", "ready", "")
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionStart, "request", proto.ExecutionStartPayload{Handle: ready.Handle, ExecutorID: ready.ExecutorID, RunID: "run", Input: proto.TextInput("input")})); err != nil {
+			t.Fatal(err)
+		}
+		synctest.Wait()
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"})); err != nil {
+			t.Fatal(err)
+		}
+		synctest.Wait()
+		time.Sleep(proto.CancellationConfirmationTimeout)
+		synctest.Wait()
+		acks := cancellationAcks(sender)
+		if len(acks) != 1 || acks[0].Applied || acks[0].Outcome != nil || (acks[0].ErrorCode != "cancel_timeout" && acks[0].ErrorCode != "cancel_failed") {
+			t.Fatalf("timeout lost unknown outcome: %+v", acks)
+		}
+		if owner.turn.cancels.Load() != 1 || owner.turn.cancelBudget != 30*time.Second || owner.closeBudget != 10*time.Second {
+			t.Fatalf("calls=%d native budget=%v close budget=%v", owner.turn.cancels.Load(), owner.turn.cancelBudget, owner.closeBudget)
+		}
+		if r.ActiveRuns() != 0 {
+			t.Fatal("confirmed close did not release retired owner")
+		}
+	})
+}

apps/daemon/internal/dispatch/executor.go

@@ -350,7 +350,7 @@ func (r *Router) closeExecutor(owner *executorState) error {
 	r.mu.Unlock()
 	var err error
 	if native != nil {
-		ctx, cancel := context.WithTimeout(context.Background(), preparedCancelTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		err = native.Close(ctx)
 		cancel()
 	}

apps/daemon/internal/dispatch/preparation_cancel.go

@@ -7,11 +7,9 @@ import (
 	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
 )
 
-const preparedCancelTimeout = 10 * time.Second
-
 func (r *Router) sendPreparedCancellation(state *sessionState, handoff *preparedHandoff, release *preparedRelease, attempt *preparedReleaseAttempt, env proto.Envelope, deliveryID string) {
 	defer r.shutdownWG.Done()
-	timer := time.NewTimer(preparedCancelTimeout)
+	timer := time.NewTimer(proto.CancellationConfirmationTimeout)
 	defer timer.Stop()
 	ack := proto.InteractionDecisionAckPayload{ErrorCode: "cancel_timeout"}
 	select {

apps/daemon/internal/dispatch/preparation_cancel_test.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/dispatch"
@@ -73,69 +74,77 @@ func (s cancellationOutputSender) Send(ctx context.Context, env proto.Envelope)
 }
 
 func TestPreparedCancellationWaitsForOutputAndCleanup(t *testing.T) {
-	sender := cancellationOutputSender{recSender: &recSender{}, entered: make(chan struct{}), release: make(chan struct{})}
-	startEntered, startReturn := make(chan struct{}), make(chan struct{})
-	cancelEntered := make(chan struct{})
-	cleanupEntered, cleanupReturn := make(chan struct{}), make(chan struct{})
-	p := &cancellationPreparation{controlledPreparation: &controlledPreparation{closed: make(chan struct{})}, outcome: proto.DonePayload{
-		Usage: proto.Usage{Tokens: &proto.TokenUsage{InputTokens: 7, OutputTokens: 3, TotalTokens: 10}}, Metadata: map[string]any{proto.DoneMetaAgentSessionID: "observed-native"},
-	}}
-	var session *fakeSession
-	p.start = func(_ context.Context, id string, _ proto.MessageInput, out chan<- proto.Envelope) (fixtureSession, error) {
-		session = &fakeSession{out: out, closeOutOnCancel: true,
-			postCancelEnvelopes: []proto.Envelope{mustEnv(t, proto.TypeDone, id, p.outcome)}}
-		out <- mustEnv(t, proto.TypeDelta, id, proto.DeltaPayload{Delta: "observed"})
-		out <- mustEnv(t, proto.TypeUsage, id, proto.UsagePayload{Usage: p.outcome.Usage})
-		close(startEntered)
-		<-startReturn
-		return session, nil
-	}
-	p.cancel = func(ctx context.Context) error {
-		close(cancelEntered)
-		if err := session.Cancel(ctx); err != nil {
-			return err
+	synctest.Test(t, func(t *testing.T) {
+		sender := cancellationOutputSender{recSender: &recSender{}, entered: make(chan struct{}), release: make(chan struct{})}
+		startEntered, startReturn := make(chan struct{}), make(chan struct{})
+		cancelEntered := make(chan struct{})
+		cleanupEntered, cleanupReturn := make(chan struct{}), make(chan struct{})
+		p := &cancellationPreparation{controlledPreparation: &controlledPreparation{closed: make(chan struct{})}, outcome: proto.DonePayload{
+			Usage: proto.Usage{Tokens: &proto.TokenUsage{InputTokens: 7, OutputTokens: 3, TotalTokens: 10}}, Metadata: map[string]any{proto.DoneMetaAgentSessionID: "observed-native"},
+		}}
+		var session *fakeSession
+		p.start = func(_ context.Context, id string, _ proto.MessageInput, out chan<- proto.Envelope) (fixtureSession, error) {
+			session = &fakeSession{out: out, closeOutOnCancel: true,
+				postCancelEnvelopes: []proto.Envelope{mustEnv(t, proto.TypeDone, id, p.outcome)}}
+			out <- mustEnv(t, proto.TypeDelta, id, proto.DeltaPayload{Delta: "observed"})
+			out <- mustEnv(t, proto.TypeUsage, id, proto.UsagePayload{Usage: p.outcome.Usage})
+			close(startEntered)
+			<-startReturn
+			return session, nil
 		}
-		close(cleanupEntered)
-		<-cleanupReturn
-		return nil
-	}
-	r := preparationRouter(t, sender, time.Minute, func(context.Context, proto.PromptRequestPayload) (preparedFixture, error) { return p, nil })
-	startCancellationPreparation(t, r, sender.recSender)
-	<-startEntered
-	if err := r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"})); err != nil {
-		t.Fatal(err)
-	}
-	<-cancelEntered
-	if len(cancellationAcks(sender.recSender)) != 0 {
-		t.Fatal("receipt preceded Start handoff")
-	}
-	close(startReturn)
-	<-sender.entered
-	if len(cancellationAcks(sender.recSender)) != 0 {
-		t.Fatal("receipt preceded output delivery")
-	}
-	// Another receipt may wait, but it must not repeat native cancellation.
-	_ = r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "second-delivery"}))
-	close(sender.release)
-	<-cleanupEntered
-	if len(cancellationAcks(sender.recSender)) != 0 || r.ActiveRuns() != 1 {
-		t.Fatal("cleanup released ownership early")
-	}
-	close(cleanupReturn)
-	waitFor(t, func() bool { return len(cancellationAcks(sender.recSender)) == 2 && r.ActiveRuns() == 0 }, "prepared cancellation receipts")
-	if p.calls.Load() != 1 || session.cancels() != 1 {
-		t.Fatal("native cancellation was repeated")
-	}
-	for _, ack := range cancellationAcks(sender.recSender) {
-		if !ack.Applied || ack.ErrorCode != "" || ack.Outcome == nil || !reflect.DeepEqual(*ack.Outcome, p.outcome) {
-			t.Fatalf("observed outcome lost: %+v", ack)
+		p.cancel = func(ctx context.Context) error {
+			close(cancelEntered)
+			if err := session.Cancel(ctx); err != nil {
+				return err
+			}
+			close(cleanupEntered)
+			<-cleanupReturn
+			return nil
 		}
-	}
-	got := sender.typesFor("run")
-	want := []string{proto.TypeDelta, proto.TypeUsage, proto.TypeDone, proto.TypeInteractionDecisionAck, proto.TypeInteractionDecisionAck}
-	if !reflect.DeepEqual(got, want) {
-		t.Fatalf("output/receipt order = %v", got)
-	}
+		r := preparationRouter(t, sender, time.Minute, func(context.Context, proto.PromptRequestPayload) (preparedFixture, error) { return p, nil })
+		startCancellationPreparation(t, r, sender.recSender)
+		<-startEntered
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"})); err != nil {
+			t.Fatal(err)
+		}
+		<-cancelEntered
+		if len(cancellationAcks(sender.recSender)) != 0 {
+			t.Fatal("receipt preceded Start handoff")
+		}
+		close(startReturn)
+		<-sender.entered
+		if len(cancellationAcks(sender.recSender)) != 0 {
+			t.Fatal("receipt preceded output delivery")
+		}
+		// Another receipt may wait, but it must not repeat native cancellation.
+		_ = r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "second-delivery"}))
+		// Confirmation remains pending past both former native and Core bounds.
+		time.Sleep(17 * time.Second)
+		synctest.Wait()
+		if len(cancellationAcks(sender.recSender)) != 0 {
+			t.Fatal("receipt expired before the shared cancellation budget")
+		}
+		close(sender.release)
+		<-cleanupEntered
+		if len(cancellationAcks(sender.recSender)) != 0 || r.ActiveRuns() != 1 {
+			t.Fatal("cleanup released ownership early")
+		}
+		close(cleanupReturn)
+		waitFor(t, func() bool { return len(cancellationAcks(sender.recSender)) == 2 && r.ActiveRuns() == 0 }, "prepared cancellation receipts")
+		if p.calls.Load() != 1 || session.cancels() != 1 {
+			t.Fatal("native cancellation was repeated")
+		}
+		for _, ack := range cancellationAcks(sender.recSender) {
+			if !ack.Applied || ack.ErrorCode != "" || ack.Outcome == nil || !reflect.DeepEqual(*ack.Outcome, p.outcome) {
+				t.Fatalf("observed outcome lost: %+v", ack)
+			}
+		}
+		got := sender.typesFor("run")
+		want := []string{proto.TypeDelta, proto.TypeUsage, proto.TypeDone, proto.TypeInteractionDecisionAck, proto.TypeInteractionDecisionAck}
+		if !reflect.DeepEqual(got, want) {
+			t.Fatalf("output/receipt order = %v", got)
+		}
+	})
 }
 
 func TestPreparedCancellationBeforeTransferPreservesUnknownOutcome(t *testing.T) {
@@ -250,44 +259,46 @@ func TestPreparedCancellationFailuresRemainConservative(t *testing.T) {
 }
 
 func TestPreparedCancellationTimeoutKeepsCapacityUntilStartReturns(t *testing.T) {
-	sender := &recSender{}
-	entered, allowReturn := make(chan struct{}), make(chan struct{})
-	p := &cancellationPreparation{controlledPreparation: &controlledPreparation{closed: make(chan struct{})}}
-	p.cancel = func(context.Context) error { return p.Close() }
-	p.start = func(ctx context.Context, _ string, _ proto.MessageInput, _ chan<- proto.Envelope) (fixtureSession, error) {
-		close(entered)
-		<-allowReturn
-		return nil, ctx.Err()
-	}
-	var count atomic.Int32
-	r := preparationRouter(t, sender, time.Minute, func(context.Context, proto.PromptRequestPayload) (preparedFixture, error) {
-		if count.Add(1) == 1 {
-			return p, nil
+	synctest.Test(t, func(t *testing.T) {
+		sender := &recSender{}
+		entered, allowReturn := make(chan struct{}), make(chan struct{})
+		p := &cancellationPreparation{controlledPreparation: &controlledPreparation{closed: make(chan struct{})}}
+		p.cancel = func(context.Context) error { return p.Close() }
+		p.start = func(ctx context.Context, _ string, _ proto.MessageInput, _ chan<- proto.Envelope) (fixtureSession, error) {
+			close(entered)
+			<-allowReturn
+			return nil, ctx.Err()
 		}
-		return &controlledPreparation{closed: make(chan struct{})}, nil
-	})
-	startCancellationPreparation(t, r, sender)
-	<-entered
+		var count atomic.Int32
+		r := preparationRouter(t, sender, time.Minute, func(context.Context, proto.PromptRequestPayload) (preparedFixture, error) {
+			if count.Add(1) == 1 {
+				return p, nil
+			}
+			return &controlledPreparation{closed: make(chan struct{})}, nil
+		})
+		startCancellationPreparation(t, r, sender)
+		<-entered
 
-	_ = r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"}))
-	deadline := time.Now().Add(12 * time.Second)
-	for len(cancellationAcks(sender)) == 0 && time.Now().Before(deadline) {
-		time.Sleep(time.Millisecond)
-	}
-	acks := cancellationAcks(sender)
-	if len(acks) != 1 || acks[0].Applied || acks[0].Outcome != nil || acks[0].ErrorCode != "cancel_timeout" || r.ActiveRuns() != 1 {
-		t.Fatal("timeout claimed settlement or lost ownership", acks)
-	}
-	if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "overflow", preparationRequest())); err == nil {
-		t.Fatal("timed-out cancellation returned capacity early")
-	}
-	close(allowReturn)
-	waitFor(t, func() bool { return r.ActiveRuns() == 0 }, "late cancelled Start cleanup")
-	if len(cancellationAcks(sender)) != 1 {
-		t.Fatal("late settlement emitted a second receipt")
-	}
-	if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "replacement", preparationRequest())); err != nil {
-		t.Fatal("completed cleanup retained capacity", err)
-	}
-	waitPreparationStatus(t, sender, "replacement", "ready", "")
+		_ = r.Handle(t.Context(), mustEnv(t, proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"}))
+		deadline := time.Now().Add(proto.CancellationConfirmationTimeout + time.Second)
+		for len(cancellationAcks(sender)) == 0 && time.Now().Before(deadline) {
+			time.Sleep(time.Millisecond)
+		}
+		acks := cancellationAcks(sender)
+		if len(acks) != 1 || acks[0].Applied || acks[0].Outcome != nil || acks[0].ErrorCode != "cancel_timeout" || r.ActiveRuns() != 1 {
+			t.Fatal("timeout claimed settlement or lost ownership", acks)
+		}
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "overflow", preparationRequest())); err == nil {
+			t.Fatal("timed-out cancellation returned capacity early")
+		}
+		close(allowReturn)
+		waitFor(t, func() bool { return r.ActiveRuns() == 0 }, "late cancelled Start cleanup")
+		if len(cancellationAcks(sender)) != 1 {
+			t.Fatal("late settlement emitted a second receipt")
+		}
+		if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "replacement", preparationRequest())); err != nil {
+			t.Fatal("completed cleanup retained capacity", err)
+		}
+		waitPreparationStatus(t, sender, "replacement", "ready", "")
+	})
 }

apps/daemon/internal/dispatch/prepared_handoff.go

@@ -160,7 +160,7 @@ func (r *Router) runPreparedRelease(state *sessionState, handoff *preparedHandof
 		outcome = &proto.DonePayload{}
 	}
 	if turn != nil {
-		ctx, cancel := context.WithTimeout(context.Background(), preparedCancelTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), proto.CancellationConfirmationTimeout)
 		cancelDone := make(chan error, 1)
 		settled := make(chan struct{})
 		go func() {

contracts/agents-api/harness-onboarding.md

@@ -118,7 +118,7 @@ A Session owns one reusable Executor in its connected Runtime; a Turn owns one i
 - For a cancelled stdio MCP call, capture its declared service before native abort can finish the local request, retain starts admitted during that Turn's cancellation, and drain the native Turn before calling `ViewSession.StopMCP` with those service labels. The host stops the selected services and all their descendants, including their earlier background work, and confirms each old Process scope with `ScopeClosed`. It fences new alias starts during this operation; an unconfirmed stop retains ownership and blocks new starts for those aliases. Other MCP services and the workspace remain. HTTP MCP is outside this operation. Stdio MCP admission requires the Process service's delegated cgroup v2 scope; a node or container without that capability returns typed unsupported.
 - After confirmed scope termination, confirm that the selected old native MCP clients cannot be reused and that subsequent calls create fresh connections through the maintained native owner before declaring the Executor reusable. A local MCP cancellation error, notification, process leader exit or attachment-close acknowledgment does not prove remote scope termination. Repeated or late cancellation joins the original Turn's recorded targets and result; it never selects a successor's services.
 
-**What the Runtime does around a Turn.** One output consumer starts before native Start, drains the bounded 64-frame channel and keeps the terminal observation until Start publication, Turn settlement and admitted operation receipts finish. Natural completion never calls Cancel. Input and function admission close before settlement; operations already admitted hold their barrier through native receipts and outbound acknowledgement. The Runtime sends cancellation to the Turn before waiting on that barrier, because a written input may need a native interrupt to produce its receipt. It joins native settlement, any required confirmed Executor close, output drain and all admitted operations before an applied acknowledgement or reuse, and only then forwards Done or an applied cancellation receipt. A failed Close can report failure while keeping the same Run and outstanding operations for retry; a closed caller wait cannot manufacture an applied input receipt. The Runtime commits native continuity and releases the old Run's admission before publishing Done, since the receiver may start another Turn at once; a late terminal-send failure belongs to the old Run and cannot invalidate a successor that already owns the Executor. Connection shutdown owns transport-loss cleanup. The settlement wait is ten seconds and the receipt send budget five seconds; a timeout is not proof of quiescence.
+**What the Runtime does around a Turn.** One output consumer starts before native Start, drains the bounded 64-frame channel and keeps the terminal observation until Start publication, Turn settlement and admitted operation receipts finish. Natural completion never calls Cancel. Input and function admission close before settlement; operations already admitted hold their barrier through native receipts and outbound acknowledgement. The Runtime sends cancellation to the Turn before waiting on that barrier, because a written input may need a native interrupt to produce its receipt. It joins native settlement, any required confirmed Executor close, output drain and all admitted operations before an applied acknowledgement or reuse, and only then forwards Done or an applied cancellation receipt. A failed Close can report failure while keeping the same Run and outstanding operations for retry; a closed caller wait cannot manufacture an applied input receipt. The Runtime commits native continuity and releases the old Run's admission before publishing Done, since the receiver may start another Turn at once; a late terminal-send failure belongs to the old Run and cannot invalidate a successor that already owns the Executor. Connection shutdown owns transport-loss cleanup. Settlement and cancellation receipt waits follow the [Core–Runtime cancellation budget](../../docs/runtime-protocol.md#active-input-receipts); the receipt send budget is five seconds; a timeout is not proof of quiescence.
 
 ## Events, inputs and optional capabilities