Skip to content

Commit 77f74c8

Browse files
committed
refactor: declare cancellation outcomes on every session owner
1 parent 5f5b4f8 commit 77f74c8

10 files changed

Lines changed: 56 additions & 41 deletions

File tree

‎apps/daemon/internal/agent/claudesdk/cancellation_test.go‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,11 +41,7 @@ func TestCancellationWaitsForDrainAndPublishesOutcome(t *testing.T) {
4141
if !errors.Is(err, context.DeadlineExceeded) {
4242
t.Fatalf("unsettled cancellation reported %v", err)
4343
}
44-
provider, ok := running.(interface{ CancellationOutcome() proto.DonePayload })
45-
if !ok {
46-
t.Fatal("missing cancellation outcome provider")
47-
}
48-
if got := provider.CancellationOutcome(); !reflect.DeepEqual(got, proto.DonePayload{}) {
44+
if got := running.CancellationOutcome(); !reflect.DeepEqual(got, proto.DonePayload{}) {
4945
t.Fatal("unsettled outcome was exposed", got)
5046
}
5147
if err := running.(*session).Steer(ctx, proto.PromptSteerPayload{InputID: "later", Input: proto.TextInput("later")}); !errors.Is(err, agent.ErrSteeringInactive) {
@@ -62,7 +58,7 @@ func TestCancellationWaitsForDrainAndPublishesOutcome(t *testing.T) {
6258
default:
6359
t.Fatal("successful cancellation preceded owned process release")
6460
}
65-
got := provider.CancellationOutcome()
61+
got := running.CancellationOutcome()
6662
if got.Content != "partialtaildrained" || got.Metadata[proto.DoneMetaAgentSessionID] != "native-session" || got.Usage.Raw["claude_sdk_result"] == nil || got.Usage.Tokens != nil {
6763
t.Fatalf("lost drained cancellation outcome: %+v", got)
6864
}

‎apps/daemon/internal/agent/harness.go‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,6 @@ type Executor interface {
110110
type Turn interface {
111111
Session
112112
DurableSteerer
113-
CancellationOutcome() proto.DonePayload
114113
// Success confirms closed output and settled native input, function,
115114
// interaction and child-work obligations. Errors cannot prove cancellation.
116115
AwaitSettlement(context.Context) (TurnSettlement, error)
@@ -123,13 +122,19 @@ type TurnSettlement struct {
123122
Reason string
124123
}
125124

126-
// Session is the cancellation surface shared by direct prompt runs and Turns.
125+
// Session is the cancellation and outcome surface shared by direct prompt runs
126+
// and Turns. Every owner exposes observed state, including direct-call sessions.
127127
// For Executor-owned Turns, AwaitSettlement and Executor.Close define settlement
128128
// and resource retirement; Cancel alone does not transfer resource ownership.
129129
type Session interface {
130130
// Cancel signals the session to abort. Idempotent. Actual teardown
131131
// happens asynchronously and is signalled via the out channel close.
132132
Cancel(ctx context.Context) error
133+
// CancellationOutcome snapshots observed native identity, usage and output.
134+
// It remains readable after Cancel; missing evidence stays unset. An empty
135+
// result means no observed evidence, not unsupported cancellation or success.
136+
// Reading it does not wait for or establish native settlement.
137+
CancellationOutcome() proto.DonePayload
133138
}
134139

135140
// Turn extension contracts. Every public Harness implements each interface;
@@ -215,10 +220,10 @@ type Prepared interface {
215220
// same native resource across Start. Read-only preparations need only Prepared.
216221
type PreparedCancellation interface {
217222
Prepared
223+
Session
218224
// Cancel returns after local cleanup and all output writes have stopped.
219225
// An error retains ownership so callers can retry this exact object serially.
220226
Cancel(context.Context) error
221-
CancellationOutcome() proto.DonePayload
222227
}
223228

224229
// A factory may return both a resource and an error when construction failed but

‎apps/daemon/internal/agent/registry_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@ func stubFactory(marker string) agent.Factory {
2222

2323
type stubSession struct{ marker string }
2424

25+
func (stubSession) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} }
26+
2527
func (stubSession) Cancel(context.Context) error { return nil }
2628
func (stubSession) SubmitPermission(context.Context, string, proto.PermissionDecisionPayload) error {
2729
return nil

‎apps/daemon/internal/dispatch/cancellation.go‎

Lines changed: 8 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,10 @@ import (
44
"context"
55
"errors"
66

7+
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
78
"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
89
)
910

10-
type cancellationOutcomeProvider interface {
11-
CancellationOutcome() proto.DonePayload
12-
}
13-
1411
func (r *Router) releaseCompletedSession(state *sessionState) error {
1512
r.mu.Lock()
1613
state.retain = false
@@ -35,7 +32,7 @@ func (r *Router) handlePromptCancel(ctx context.Context, env proto.Envelope) err
3532
if state != nil {
3633
state.retain = false
3734
}
38-
var cancelSession func(context.Context) error
35+
var owner agent.Session
3936
if state != nil && state.preparedHandoff != nil {
4037
handoff := state.preparedHandoff
4138
release, attempt := r.claimPreparedReleaseLocked(state, true, "", true)
@@ -47,22 +44,20 @@ func (r *Router) handlePromptCancel(ctx context.Context, env proto.Envelope) err
4744
return nil
4845
}
4946
if state != nil && state.session != nil {
50-
cancelSession = state.session.Cancel
47+
owner = state.session
5148
}
5249
r.mu.Unlock()
5350
ack := proto.InteractionDecisionAckPayload{DeliveryID: request.DeliveryID, ErrorCode: "run_inactive"}
54-
if state != nil && cancelSession == nil {
51+
if state != nil && owner == nil {
5552
ack.ErrorCode = "not_ready"
56-
} else if cancelSession != nil {
57-
if err := cancelSession(ctx); err != nil {
53+
} else if owner != nil {
54+
if err := owner.Cancel(ctx); err != nil {
5855
r.log.WarnContext(ctx, "session.Cancel failed", "run_id", env.ID, "err", err)
5956
ack.ErrorCode = "cancel_failed"
6057
} else {
6158
ack.Applied, ack.ErrorCode = true, ""
62-
if provider, ok := state.session.(cancellationOutcomeProvider); ok {
63-
outcome := provider.CancellationOutcome()
64-
ack.Outcome = &outcome
65-
}
59+
outcome := owner.CancellationOutcome()
60+
ack.Outcome = &outcome
6661
}
6762
state.ctxCancel()
6863
}

‎apps/daemon/internal/dispatch/cancellation_test.go‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import "github.com/MiniMax-AI/OpenAgentCore/internal/harnessconfig"
55
import (
66
"context"
77
"errors"
8+
"reflect"
89
"testing"
910

1011
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
@@ -17,10 +18,11 @@ type cancelReceiptSession struct {
1718
entered chan struct{}
1819
release chan struct{}
1920
err error
21+
outcome proto.DonePayload
2022
}
2123

2224
func (s *cancelReceiptSession) CancellationOutcome() proto.DonePayload {
23-
return proto.DonePayload{Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native-cancelled"}}
25+
return s.outcome
2426
}
2527

2628
func TestCompletionWaitsForNativeWriterRelease(t *testing.T) {
@@ -58,14 +60,22 @@ func (s *cancelReceiptSession) Cancel(ctx context.Context) error {
5860
}
5961

6062
func TestCancellationReceiptFollowsAdapterOutcome(t *testing.T) {
61-
for _, fails := range []bool{false, true} {
62-
t.Run(map[bool]string{false: "applied", true: "rejected"}[fails], func(t *testing.T) {
63+
observed := proto.DonePayload{Content: "partial output", Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native-cancelled"}}
64+
for _, test := range []struct {
65+
name string
66+
outcome proto.DonePayload
67+
err error
68+
}{
69+
{name: "observed", outcome: observed},
70+
{name: "unknown"},
71+
{name: "failed", outcome: observed, err: errors.New("adapter could not cancel")},
72+
{name: "unsupported", outcome: observed, err: agent.ErrUnsupportedOperation},
73+
{name: "deadline", outcome: observed, err: context.DeadlineExceeded},
74+
} {
75+
t.Run(test.name, func(t *testing.T) {
6376
h := newHarness(t)
6477
defer h.router.Shutdown(context.Background())
65-
sess := &cancelReceiptSession{entered: make(chan struct{}), release: make(chan struct{})}
66-
if fails {
67-
sess.err = errors.New("adapter could not cancel")
68-
}
78+
sess := &cancelReceiptSession{entered: make(chan struct{}), release: make(chan struct{}), outcome: test.outcome, err: test.err}
6979
h.reg.RegisterKind(proto.SupportedAgentKind{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{})}, harnessconfig.Configuration{}, func(ctx context.Context, req proto.PromptRequestPayload, out chan<- proto.Envelope) (agent.Session, error) {
7080
sess.fakeSession = &fakeSession{out: out, closeOutOnCancel: true}
7181
return sess, nil
@@ -93,11 +103,15 @@ func TestCancellationReceiptFollowsAdapterOutcome(t *testing.T) {
93103
found = true
94104
var ack proto.InteractionDecisionAckPayload
95105
_ = env.DecodePayload(&ack)
96-
if ack.Applied == fails || ack.DeliveryID != "cancel-1" {
106+
if ack.Applied != (test.err == nil) || ack.DeliveryID != "cancel-1" {
97107
t.Fatalf("wrong receipt: %+v", ack)
98108
}
99-
if !fails && (ack.Outcome == nil || ack.Outcome.Metadata[proto.DoneMetaAgentSessionID] != "native-cancelled") {
100-
t.Fatal("cancellation receipt lost native identity")
109+
if test.err == nil {
110+
if ack.ErrorCode != "" || ack.Outcome == nil || !reflect.DeepEqual(*ack.Outcome, test.outcome) {
111+
t.Fatalf("cancellation receipt changed observed evidence: %+v", ack)
112+
}
113+
} else if ack.ErrorCode != "cancel_failed" || ack.Outcome != nil {
114+
t.Fatalf("failed cancellation supplied a success outcome: %+v", ack)
101115
}
102116
}
103117
}

‎apps/daemon/internal/dispatch/optional_interactions_test.go‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,8 @@ import (
1111
"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto/prototest"
1212
)
1313

14-
// Wrapping only Cancel proves that no responder stubs are required for a Session.
15-
type lifecycleOnly struct{ cancel func(context.Context) error }
16-
17-
func (s lifecycleOnly) Cancel(ctx context.Context) error { return s.cancel(ctx) }
14+
// Wrapping Session proves that no responder stubs are required for a Session.
15+
type lifecycleOnly struct{ agent.Session }
1816

1917
func TestOptionalInteractionResponders(t *testing.T) {
2018
for _, ask := range []bool{false, true} {
@@ -25,7 +23,7 @@ func TestOptionalInteractionResponders(t *testing.T) {
2523
h.reg.RegisterKind(proto.SupportedAgentKind{Kind: "minimal", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{})}, harnessconfig.Configuration{}, func(_ context.Context, _ proto.PromptRequestPayload, out chan<- proto.Envelope) (agent.Session, error) {
2624
output = out
2725
s := &fakeSession{out: out, closeOutOnCancel: true}
28-
return lifecycleOnly{cancel: s.Cancel}, nil
26+
return lifecycleOnly{Session: s}, nil
2927
})
3028
if err := h.router.Handle(t.Context(), mustEnv(t, proto.TypePromptRequest, "run", proto.PromptRequestPayload{AgentKind: "minimal"})); err != nil {
3129
t.Fatal(err)

‎apps/daemon/internal/dispatch/preparation_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,8 +65,8 @@ func (p *controlledPreparation) CancellationOutcome() proto.DonePayload {
6565
p.mu.Lock()
6666
session := p.session
6767
p.mu.Unlock()
68-
if provider, ok := session.(interface{ CancellationOutcome() proto.DonePayload }); ok {
69-
return provider.CancellationOutcome()
68+
if session != nil {
69+
return session.CancellationOutcome()
7070
}
7171
return proto.DonePayload{}
7272
}

‎apps/daemon/internal/dispatch/router_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,8 @@ type askCall struct {
8484
decision proto.PromptForUserChoiceDecisionPayload
8585
}
8686

87+
func (s *fakeSession) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} }
88+
8789
func (s *fakeSession) Cancel(context.Context) error {
8890
s.cancelMu.Lock()
8991
s.cancelCalls++

‎apps/daemon/internal/dispatch/suspend_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ func (s suspendSender) Send(ctx context.Context, env proto.Envelope) error { ret
1717

1818
type suspendedSession struct{ cancelled atomic.Int32 }
1919

20+
func (s *suspendedSession) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} }
21+
2022
func (s *suspendedSession) Cancel(context.Context) error { s.cancelled.Add(1); return nil }
2123

2224
func suspensionRouter(t *testing.T, sender Sender) *Router {

‎contracts/agents-api/harness-onboarding.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ For example, the Codex adapter keeps its app-server and thread, the Claude adapt
6363
| Interface or contract | Required handling | Obligation |
6464
| --- | --- | --- |
6565
| `ExecutorFactory`, `Executor.StartTurn`, `Executor.Close` | Real implementation | Prepare without model input; keep ownership of failed or uncertain resources; confirm cleanup |
66-
| `Turn`, `Session.Cancel`, `CancellationOutcome`, `AwaitSettlement` | Real implementation | Cancel the exact Turn, keep observed results and confirm settlement independently of cancellation requests |
66+
| `Session`, `Turn`, `CancellationOutcome`, `AwaitSettlement` | Real implementation | Cancel the exact Turn, keep observed results and confirm settlement independently of cancellation requests |
6767
| `DurableSteerer` | Real implementation on every Turn | Distinguish a complete write from the native application receipt; keep retry identity |
6868
| `Steerer` | Explicit implementation or Unsupported | Additional non-durable active-Turn input |
6969
| `FunctionResultSubmitter` | Explicit implementation or Unsupported | Match native call and result identity and acknowledge application |
@@ -112,7 +112,8 @@ A Session owns one reusable Executor in its connected Runtime; a Turn owns one i
112112
- An error means settlement is unconfirmed and frees neither ownership nor capacity. Caller deadlines stop the wait, not the tracked cleanup. Retry the same cleanup target serially; a failed cleanup blocks replacement and keeps its resource slot.
113113
- `Executor.Close` confirms resource retirement independently of the Turn outcome: an immutable Turn error must not prevent closing the native transport once its work and output have stopped.
114114
- Include owned background work in settlement and keep the exact native cleanup target after a failure. Native termination belongs to the adapter; a bulk cleanup acknowledgement alone does not establish quiescence.
115-
- The observed cancellation outcome keeps native identity, Usage and output without fabricating missing evidence.
115+
- Every `Session`, including a direct-call factory result, declares `CancellationOutcome`. `Turn` and `PreparedCancellation` inherit it. The snapshot keeps observed native identity, Usage and output and remains readable after cancellation. Missing evidence stays unset; an empty `DonePayload` means nothing has been observed, not that cancellation succeeded or is unsupported. Reading the snapshot does not wait for settlement.
116+
- Direct-call `Session.Cancel` requests cancellation; output closure signals teardown. Executable `PreparedCancellation.Cancel` waits for local cleanup and output writes to stop. Turn settlement still requires `AwaitSettlement` and any required `Executor.Close`; neither a successful cancellation request nor its snapshot replaces those waits.
116117

117118
**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, function and interaction 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.
118119

0 commit comments

Comments
 (0)