Make Turn one interface

apps/daemon/internal/agent/claudesdk/cancellation.go

@@ -3,7 +3,6 @@ package claudesdk
 import (
 	"context"
 	"errors"
-	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
 
 	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
 )
@@ -63,5 +62,3 @@ func (s *session) CancellationOutcome() proto.DonePayload {
 		return proto.DonePayload{}
 	}
 }
-
-var _ agent.Turn = (*session)(nil)

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

@@ -44,7 +44,7 @@ func TestCancellationWaitsForDrainAndPublishesOutcome(t *testing.T) {
 	if got := running.CancellationOutcome(); !reflect.DeepEqual(got, proto.DonePayload{}) {
 		t.Fatal("unsettled outcome was exposed", got)
 	}
-	if err := running.(*session).Steer(ctx, proto.PromptSteerPayload{InputID: "later", Input: proto.TextInput("later")}); !errors.Is(err, agent.ErrSteeringInactive) {
+	if err := running.SteerWithReceipt(ctx, proto.PromptSteerPayload{InputID: "later", Input: proto.TextInput("later")}, func() {}); !errors.Is(err, agent.ErrSteeringInactive) {
 		t.Fatal("cancelled execution accepted steering", err)
 	}
 	if err := os.WriteFile(filepath.Join(config.StateDir, "release"), nil, 0o600); err != nil {

apps/daemon/internal/agent/claudesdk/contracts.go

@@ -2,13 +2,7 @@ package claudesdk
 
 import "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
 
-// Every public Harness implements each small contract explicitly. Unsupported
-// extensions return agent.ErrUnsupportedOperation without native effects.
 var (
-	_ agent.Executor                = (*executor)(nil)
-	_ agent.Turn                    = (*session)(nil)
-	_ agent.Session                 = (*session)(nil)
-	_ agent.DurableSteerer          = (*session)(nil)
-	_ agent.Steerer                 = (*session)(nil)
-	_ agent.FunctionResultSubmitter = (*session)(nil)
+	_ agent.Executor = (*executor)(nil)
+	_ agent.Turn     = (*session)(nil)
 )

apps/daemon/internal/agent/claudesdk/executor.go

@@ -300,9 +300,6 @@ func (s *session) AwaitSettlement(ctx context.Context) (agent.TurnSettlement, er
 	}
 }
 
-var _ agent.Executor = (*executor)(nil)
-var _ agent.Turn = (*session)(nil)
-
 // A timed-out operation retains its original Turn identity. Its callback cannot
 // retire a healthy successor after that operation has otherwise settled.
 func (s *session) invalidate() {

apps/daemon/internal/agent/claudesdk/executor_confirmation_test.go

@@ -38,7 +38,7 @@ func TestExecutorNativeConfirmationSurvivesCleanup(t *testing.T) {
 			if mode == "pending_input" {
 				inputReceipt = make(chan error, 1)
 				go func() {
-					inputReceipt <- turn.(*session).SteerWithReceipt(t.Context(), proto.PromptSteerPayload{InputID: "input", Input: proto.TextInput("extra")}, nil)
+					inputReceipt <- turn.SteerWithReceipt(t.Context(), proto.PromptSteerPayload{InputID: "input", Input: proto.TextInput("extra")}, func() {})
 				}()
 				if event := <-out; event.Type != proto.TypeDelta {
 					t.Fatal("input write barrier missing")

apps/daemon/internal/agent/claudesdk/functions_test.go

@@ -12,7 +12,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
 	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
 )
 
@@ -31,7 +30,6 @@ func TestFunctionTurnNativeReceipts(t *testing.T) {
 				t.Fatal(err)
 			}
 			defer running.Cancel(context.Background())
-			submitter := running.(agent.FunctionResultSubmitter)
 			submissions := make(chan error, 2)
 			calls := 0
 			failed := false
@@ -48,17 +46,17 @@ func TestFunctionTurnNativeReceipts(t *testing.T) {
 					}
 					calls++
 					invalid := proto.FunctionResultPayload{DeliveryID: "delivery-" + call.CallID, CallID: call.CallID, Success: true}
-					if err := submitter.SubmitFunctionResult(ctx, invalid); err == nil {
+					if err := running.SubmitFunctionResult(ctx, invalid); err == nil {
 						t.Fatal("missing content consumed call")
 					}
 					image := "https://example.invalid/image"
 					invalid.Content = []proto.InputContent{{Type: "input_image", ImageURL: &image}}
-					if err := submitter.SubmitFunctionResult(ctx, invalid); err == nil {
+					if err := running.SubmitFunctionResult(ctx, invalid); err == nil {
 						t.Fatal("image should fail before delivery")
 					}
 					first, second := "first-"+call.CallID, "second-"+call.CallID
 					value := proto.FunctionResultPayload{DeliveryID: "delivery-" + call.CallID, CallID: call.CallID, Success: call.CallID == "b", Content: []proto.InputContent{{Type: "input_text", Text: &first}, {Type: "input_text", Text: &second}}}
-					go func() { submissions <- submitter.SubmitFunctionResult(ctx, value) }()
+					go func() { submissions <- running.SubmitFunctionResult(ctx, value) }()
 				case proto.TypeToolCall:
 					var tool proto.ToolCallPayload
 					if err := event.DecodePayload(&tool); err != nil {

apps/daemon/internal/agent/claudesdk/steering.go

@@ -24,15 +24,8 @@ type steeringState struct {
 	seen      map[string]bool
 }
 
-var _ agent.Steerer = (*session)(nil)
-
-// Steer waits for native consumption, which may occur in a later native turn
-// within this one SDK query. A completed stdin write is not a receipt.
-func (s *session) Steer(ctx context.Context, input proto.PromptSteerPayload) error {
-	return s.SteerWithReceipt(ctx, input, nil)
-}
-
-// SteerWithReceipt separates a complete bridge write from native consumption.
+// SteerWithReceipt separates a complete bridge write from native consumption,
+// which may occur in a later native turn within this one SDK query.
 func (s *session) SteerWithReceipt(ctx context.Context, input proto.PromptSteerPayload, written func()) error {
 	select {
 	case <-s.cancelOutput:
@@ -91,9 +84,7 @@ func (s *session) SteerWithReceipt(ctx context.Context, input proto.PromptSteerP
 		s.invalidate()
 		return fmt.Errorf("claudesdk: input transport failed")
 	}
-	if written != nil {
-		written()
-	}
+	written()
 	select {
 	case err := <-pending.receipt:
 		return err

apps/daemon/internal/agent/claudesdk/steering_test.go

@@ -67,7 +67,7 @@ func TestSteeringReceiptsAndLifecycle(t *testing.T) {
 						}
 					})
 				} else {
-					reply <- s.Steer(receiptCtx, input)
+					reply <- s.SteerWithReceipt(receiptCtx, input, func() {})
 				}
 			}()
 			if mode == "blocked-write" {
@@ -144,7 +144,7 @@ func TestSteeringReceiptsAndLifecycle(t *testing.T) {
 			if mode == "duplicate-usage" && measurements != 1 {
 				t.Fatal("duplicate measurement was published")
 			}
-			if err := s.Steer(ctx, input); !errors.Is(err, agent.ErrSteeringInactive) {
+			if err := s.SteerWithReceipt(ctx, input, func() {}); !errors.Is(err, agent.ErrSteeringInactive) {
 				t.Fatal("completed execution accepted input", err)
 			}
 			if _, err := s.AwaitSettlement(ctx); err != nil && (mode == "success" || mode == "phased" || mode == "timeout") {
@@ -159,11 +159,11 @@ func TestSteeringReceiptsAndLifecycle(t *testing.T) {
 
 func TestSteeringDoesNotSendBeforeReadiness(t *testing.T) {
 	s := &session{process: &clirunner.Process{}}
-	if err := s.Steer(context.Background(), proto.PromptSteerPayload{InputID: "one", Input: proto.TextInput("hello")}); !errors.Is(err, agent.ErrSteeringNotReady) {
+	if err := s.SteerWithReceipt(context.Background(), proto.PromptSteerPayload{InputID: "one", Input: proto.TextInput("hello")}, func() {}); !errors.Is(err, agent.ErrSteeringNotReady) {
 		t.Fatal(err)
 	}
 	s.stopSteering()
-	if err := s.Steer(context.Background(), proto.PromptSteerPayload{InputID: "one", Input: proto.TextInput("hello")}); !errors.Is(err, agent.ErrSteeringInactive) {
+	if err := s.SteerWithReceipt(context.Background(), proto.PromptSteerPayload{InputID: "one", Input: proto.TextInput("hello")}, func() {}); !errors.Is(err, agent.ErrSteeringInactive) {
 		t.Fatal(err)
 	}
 }

apps/daemon/internal/agent/codex/contracts.go

@@ -2,13 +2,7 @@ package codex
 
 import "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
 
-// Every public Harness implements each small contract explicitly. Unsupported
-// extensions return agent.ErrUnsupportedOperation without native effects.
 var (
-	_ agent.Executor                = (*Executor)(nil)
-	_ agent.Turn                    = (*Session)(nil)
-	_ agent.Session                 = (*Session)(nil)
-	_ agent.DurableSteerer          = (*Session)(nil)
-	_ agent.Steerer                 = (*Session)(nil)
-	_ agent.FunctionResultSubmitter = (*Session)(nil)
+	_ agent.Executor = (*Executor)(nil)
+	_ agent.Turn     = (*Session)(nil)
 )

apps/daemon/internal/agent/codex/executor_turn.go

@@ -139,8 +139,6 @@ func (s *Session) Cancel(ctx context.Context) error {
 	return err
 }
 
-var _ agent.Turn = (*Session)(nil)
-
 // Server callbacks retain their originating Turn, whose ID must match exactly.
 func (s *Session) onServerRequest(method string, handler ServerRequestHandler) {
 	s.rpc.OnServerRequest(method, func(raw json.RawMessage, id any) (any, error) {

apps/daemon/internal/agent/codex/session.go

@@ -107,8 +107,6 @@ type Session struct {
 	outcome cancellationOutcomeState
 }
 
-var _ agent.Session = (*Session)(nil)
-
 // ---------------------------------------------------------------------------
 // notification handlers
 // ---------------------------------------------------------------------------

apps/daemon/internal/agent/codex/session_steering.go

@@ -25,19 +25,9 @@ type TurnSteerParams struct {
 	Input          []UserInput `json:"input"`
 }
 
-var _ agent.Steerer = (*Session)(nil)
-
-// Steer returns success only after Codex accepts input for this native turn.
-func (s *Session) Steer(ctx context.Context, input proto.PromptSteerPayload) error {
-	return s.steer(ctx, input, nil)
-}
-
-// SteerWithReceipt waits under the Run context after reporting the complete write.
+// SteerWithReceipt returns success only after Codex accepts input for this
+// native turn. It waits under the Run context after reporting the complete write.
 func (s *Session) SteerWithReceipt(ctx context.Context, input proto.PromptSteerPayload, written func()) error {
-	return s.steer(ctx, input, written)
-}
-
-func (s *Session) steer(ctx context.Context, input proto.PromptSteerPayload, written func()) error {
 	if !s.beginOperation() {
 		return agent.ErrSteeringInactive
 	}
@@ -67,19 +57,13 @@ func (s *Session) steer(ctx context.Context, input proto.PromptSteerPayload, wri
 		return agent.ErrSteeringRejected
 	}
 	params := TurnSteerParams{ThreadID: threadID, ExpectedTurnID: turnID, Input: content}
-	timeout := s.rpc.cfg.RequestTimeout
-	if written != nil {
-		timeout = 0
-	}
 	raw, err := s.rpc.requestWithTimeout(ctx, "turn/steer", params, func(frame any) error {
 		if err := s.rpc.writeFrameContext(ctx, frame); err != nil {
 			return err
 		}
-		if written != nil {
-			written()
-		}
+		written()
 		return nil
-	}, timeout, nil)
+	}, 0, nil)
 	if err != nil {
 		var rejected *JsonRpcError
 		if errors.As(err, &rejected) {

apps/daemon/internal/agent/codex/session_steering_lifecycle_test.go

@@ -25,9 +25,9 @@ func TestSteeringReceiptTimeoutAndCompletionKeepProcessAlive(t *testing.T) {
 			}
 			ctx, cancel := context.WithTimeout(context.Background(), timeout)
 			defer cancel()
-			done := make(chan error, 1)
+			done, written := make(chan error, 1), make(chan struct{})
 			go func() {
-				done <- s.Steer(ctx, proto.PromptSteerPayload{InputID: "input", Input: proto.TextInput("extra")})
+				done <- s.SteerWithReceipt(ctx, proto.PromptSteerPayload{InputID: "input", Input: proto.TextInput("extra")}, func() { close(written) })
 			}()
 			var request JsonRpcRequest
 			if err := json.NewDecoder(server.FromClient).Decode(&request); err != nil {
@@ -36,19 +36,10 @@ func TestSteeringReceiptTimeoutAndCompletionKeepProcessAlive(t *testing.T) {
 			// Withhold the response after reading the entire request frame.
 			if complete {
 				// Reading the pipe does not mean its writer has returned yet.
-				for {
-					client.pendingMu.Lock()
-					pending := client.pending[request.ID]
-					waiting := pending != nil && pending.timer != nil
-					client.pendingMu.Unlock()
-					if waiting {
-						break
-					}
-					select {
-					case <-ctx.Done():
-						t.Fatal("steering request did not reach response wait")
-					case <-time.After(time.Millisecond):
-					}
+				select {
+				case <-written:
+				case <-ctx.Done():
+					t.Fatal("steering request did not reach response wait")
 				}
 				s.onTurnCompleted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn","status":"completed"}}`))
 			}
@@ -125,7 +116,7 @@ func TestBlockedSteeringWriteEndsRunWithTerminalFrames(t *testing.T) {
 	s := turn.(*Session)
 	callCtx, callCancel := context.WithTimeout(ctx, 50*time.Millisecond)
 	defer callCancel()
-	if err := s.Steer(callCtx, proto.PromptSteerPayload{InputID: "blocked", Input: proto.TextInput("extra")}); err == nil {
+	if err := s.SteerWithReceipt(callCtx, proto.PromptSteerPayload{InputID: "blocked", Input: proto.TextInput("extra")}, func() {}); err == nil {
 		t.Fatal("blocked write accepted")
 	}
 	if client.Alive() {

apps/daemon/internal/agent/codex/session_steering_receipt_test.go

@@ -10,66 +10,46 @@ import (
 )
 
 func TestDurableSteeringBypassesOnlyNativeResponseDeadline(t *testing.T) {
-	for _, durable := range []bool{false, true} {
-		t.Run(map[bool]string{false: "legacy", true: "durable"}[durable], func(t *testing.T) {
-			client, server, cleanup := NewTestClient()
-			defer cleanup()
-			client.cfg.RequestTimeout = 20 * time.Millisecond
-			s := &Session{rpc: client.JSONRPCClient, cancelCtx: context.Background()}
-			s.setThreadID("thread")
-			s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`))
-			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-			defer cancel()
-			written := make(chan struct{})
-			reply := make(chan error, 1)
-			go func() {
-				input := proto.PromptSteerPayload{InputID: "extra", Input: proto.TextInput("text")}
-				if durable {
-					reply <- s.SteerWithReceipt(ctx, input, func() { close(written) })
-				} else {
-					reply <- s.Steer(ctx, input)
-				}
-			}()
-			if durable {
-				select {
-				case <-written:
-					t.Fatal("written before frame was read")
-				default:
-				}
-			}
-			var request JsonRpcRequest
-			if err := json.NewDecoder(server.FromClient).Decode(&request); err != nil {
-				t.Fatal(err)
-			}
-			if durable {
-				select {
-				case <-written:
-				case <-ctx.Done():
-					t.Fatal("missing write phase")
-				}
-			}
-			select {
-			case err := <-reply:
-				if durable || err == nil {
-					t.Fatal("unexpected response deadline", err)
-				}
-			case <-time.After(60 * time.Millisecond):
-				if !durable {
-					t.Fatal("legacy timeout disappeared")
-				}
-			}
-			if durable {
-				if err := json.NewEncoder(server.ToClient).Encode(map[string]any{"id": request.ID, "result": map[string]string{"turnId": "turn"}}); err != nil {
-					t.Fatal(err)
-				}
-				if err := <-reply; err != nil {
-					t.Fatal(err)
-				}
-			}
-			if !client.Alive() {
-				t.Fatal("receipt wait killed process")
-			}
-		})
+	client, server, cleanup := NewTestClient()
+	defer cleanup()
+	client.cfg.RequestTimeout = 20 * time.Millisecond
+	s := &Session{rpc: client.JSONRPCClient, cancelCtx: context.Background()}
+	s.setThreadID("thread")
+	s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`))
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	written := make(chan struct{})
+	reply := make(chan error, 1)
+	go func() {
+		reply <- s.SteerWithReceipt(ctx, proto.PromptSteerPayload{InputID: "extra", Input: proto.TextInput("text")}, func() { close(written) })
+	}()
+	select {
+	case <-written:
+		t.Fatal("written before frame was read")
+	default:
+	}
+	var request JsonRpcRequest
+	if err := json.NewDecoder(server.FromClient).Decode(&request); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-written:
+	case <-ctx.Done():
+		t.Fatal("missing write phase")
+	}
+	select {
+	case err := <-reply:
+		t.Fatal("unexpected response deadline", err)
+	case <-time.After(60 * time.Millisecond):
+	}
+	if err := json.NewEncoder(server.ToClient).Encode(map[string]any{"id": request.ID, "result": map[string]string{"turnId": "turn"}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-reply; err != nil {
+		t.Fatal(err)
+	}
+	if !client.Alive() {
+		t.Fatal("receipt wait killed process")
 	}
 }