tutti/packages/agent/daemon/runtime/controller_provisional_acceptance_test.go

package agentruntime

import (
	"context"
	"errors"
	"sync"
	"testing"

	agentsessionstore "github.com/tutti-os/tutti/packages/agent/daemon/activity"
	activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
)

type rejectingProviderAcceptanceAdapter struct {
	recordingStartAdapter
	failure error
}

// providerlessTerminalAcceptanceAdapter models a provider invocation that
// reaches an authoritative canonical terminal before a provider Turn identity
// can be bound. The dispatch result remains providerless; the exact terminal
// event is the lifecycle authority.
type providerlessTerminalAcceptanceAdapter struct {
	recordingStartAdapter
	failure error
}

func (*providerlessTerminalAcceptanceAdapter) ForkCapabilities(
	context.Context,
	Session,
) (SessionForkCapabilities, error) {
	return SessionForkCapabilities{ThroughTurn: true}, nil
}

func (*providerlessTerminalAcceptanceAdapter) Fork(
	context.Context,
	SessionForkInput,
) (SessionForkResult, error) {
	return SessionForkResult{}, nil
}

func (a *providerlessTerminalAcceptanceAdapter) ExecWithProviderAcceptance(
	_ context.Context,
	session Session,
	_ []PromptContentBlock,
	_ string,
	turnID string,
	_ EventSink,
	_ CommandSnapshotSink,
	_ ProviderDispatchSink,
	_ ProviderAcceptanceBarrier,
) ([]activityshared.Event, error) {
	return []activityshared.Event{newTurnActivityEvent(
		session,
		EventTurnFailed,
		turnID,
		SessionStatusFailed,
		"",
		"",
		map[string]any{
			"error":      a.failure.Error(),
			"stopReason": "failed_before_provider_acceptance",
		},
	)}, a.failure
}

func (*providerlessTerminalAcceptanceAdapter) UsesRootProviderTurnLifecycle() bool {
	return true
}

type retryingTerminalReporter struct {
	recordingReporter
	attemptMu                sync.Mutex
	terminalAttempts         int
	commitBeforeFirstFailure bool
}

func (r *retryingTerminalReporter) Report(
	ctx context.Context,
	report agentsessionstore.ReportActivityInput,
) error {
	terminal := false
	for _, patch := range report.StatePatches {
		if patch.Turn != nil && patch.Turn.Phase == "settled" {
			terminal = true
			break
		}
	}
	if !terminal {
		return r.recordingReporter.Report(ctx, report)
	}

	r.attemptMu.Lock()
	r.terminalAttempts++
	attempt := r.terminalAttempts
	commitBeforeFailure := attempt == 1 && r.commitBeforeFirstFailure
	r.attemptMu.Unlock()
	if commitBeforeFailure {
		_ = r.recordingReporter.Report(ctx, report)
	}
	if attempt == 1 {
		return errors.New("terminal report unavailable")
	}
	return r.recordingReporter.Report(ctx, report)
}

func (r *retryingTerminalReporter) attempts() int {
	r.attemptMu.Lock()
	defer r.attemptMu.Unlock()
	return r.terminalAttempts
}

func (*rejectingProviderAcceptanceAdapter) ForkCapabilities(
	context.Context,
	Session,
) (SessionForkCapabilities, error) {
	return SessionForkCapabilities{ThroughTurn: true}, nil
}

func (*rejectingProviderAcceptanceAdapter) Fork(
	context.Context,
	SessionForkInput,
) (SessionForkResult, error) {
	return SessionForkResult{}, nil
}

func (a *rejectingProviderAcceptanceAdapter) ExecWithProviderAcceptance(
	_ context.Context,
	_ Session,
	_ []PromptContentBlock,
	_ string,
	_ string,
	_ EventSink,
	_ CommandSnapshotSink,
	reportDispatch ProviderDispatchSink,
	_ ProviderAcceptanceBarrier,
) ([]activityshared.Event, error) {
	reportDispatch(ProviderDispatchResult{
		Disposition: DispatchDispositionRejected,
		Failure:     a.failure,
	})
	return nil, a.failure
}

func (*rejectingProviderAcceptanceAdapter) UsesRootProviderTurnLifecycle() bool {
	return true
}

func TestControllerProvisionalSessionPublishesPromptAndSettlesRejectedFirstTurn(t *testing.T) {
	t.Parallel()

	providerFailure := &AppError{
		Code:    "auth_required",
		Message: "Claude Code needs authentication",
	}
	adapter := &rejectingProviderAcceptanceAdapter{
		recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
		failure:               providerFailure,
	}
	reporter := &recordingReporter{}
	controller := NewController([]Adapter{adapter}, reporter)
	if _, err := controller.Start(t.Context(), StartInput{
		RoomID: "room-1", AgentSessionID: "session-rejected",
		Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
	}); err != nil {
		t.Fatal(err)
	}

	result, err := controller.Exec(t.Context(), ExecInput{
		RoomID: "room-1", AgentSessionID: "session-rejected",
		TurnID: "turn-rejected", Content: textPrompt("hello"),
		RequireProviderAcceptance: true,
	})
	var appErr *AppError
	if !errors.As(err, &appErr) || appErr.Code != "auth_required" {
		t.Fatalf("Exec() error = %#v, want auth_required AppError", err)
	}
	if result.ProviderDispatch == nil ||
		result.ProviderDispatch.Disposition != DispatchDispositionRejected ||
		result.ProviderDispatch.Acceptance != nil {
		t.Fatalf("provider dispatch = %#v, want rejected", result.ProviderDispatch)
	}
	waitForCondition(t, func() bool {
		return !controller.HasActiveTurn("room-1", "session-rejected")
	})
	sessions := controller.Sessions("room-1")
	if len(sessions) != 1 || sessions[0].AgentSessionID != "session-rejected" {
		t.Fatalf("visible sessions = %#v, want rejected session retained", sessions)
	}
	if sessions[0].Status != SessionStatusFailed {
		t.Fatalf("rejected session status = %q, want failed", sessions[0].Status)
	}

	reports := reporter.waitForCalls(t, 2)
	var foundSubmitted, foundFailed bool
	for _, report := range reports {
		for _, patch := range report.report.StatePatches {
			if patch.Turn == nil || patch.Turn.TurnID != "turn-rejected" {
				continue
			}
			if patch.Turn.Phase == "submitted" && patch.RuntimeContext["visible"] == true {
				foundSubmitted = true
			}
			if patch.Turn.Phase == "settled" && patch.Turn.Outcome == "failed" {
				foundFailed = true
			}
		}
	}
	if !foundSubmitted || !foundFailed {
		t.Fatalf("reports = %#v, want visible submitted and failed Turn reports", reports)
	}
}

func TestControllerProviderlessCanonicalTerminalSettlesRootTurn(t *testing.T) {
	t.Parallel()

	providerFailure := errors.New("provider failed before durable acceptance")
	adapter := &providerlessTerminalAcceptanceAdapter{
		recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
		failure:               providerFailure,
	}
	reporter := &recordingReporter{}
	controller := NewController([]Adapter{adapter}, reporter)
	if _, err := controller.Start(t.Context(), StartInput{
		RoomID: "room-1", AgentSessionID: "session-providerless-terminal",
		Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
	}); err != nil {
		t.Fatal(err)
	}

	result, err := controller.Exec(t.Context(), ExecInput{
		RoomID: "room-1", AgentSessionID: "session-providerless-terminal",
		TurnID: "turn-providerless-terminal", Content: textPrompt("hello"),
		RequireProviderAcceptance: true,
	})
	if err != nil {
		t.Fatalf("Exec() error = %v, want canonical submit retained", err)
	}
	if result.ProviderDispatch == nil ||
		result.ProviderDispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn ||
		result.ProviderDispatch.Acceptance != nil {
		t.Fatalf(
			"provider dispatch = %#v, want applied_without_provider_turn",
			result.ProviderDispatch,
		)
	}
	waitForCondition(t, func() bool {
		return !controller.HasActiveTurn("room-1", "session-providerless-terminal")
	})
	sessions := controller.Sessions("room-1")
	if len(sessions) != 1 || sessions[0].AgentSessionID != "session-providerless-terminal" {
		t.Fatalf("visible sessions = %#v, want providerless-terminal session retained", sessions)
	}
	if sessions[0].Status != SessionStatusFailed {
		t.Fatalf("providerless-terminal session status = %q, want failed", sessions[0].Status)
	}

	reports := reporter.waitForCalls(t, 2)
	var foundSubmitted, foundFailed bool
	for _, report := range reports {
		for _, patch := range report.report.StatePatches {
			if patch.Turn == nil || patch.Turn.TurnID != "turn-providerless-terminal" {
				continue
			}
			if patch.Turn.Phase == "submitted" && patch.RuntimeContext["visible"] == true {
				foundSubmitted = true
			}
			if patch.Turn.Phase == "settled" && patch.Turn.Outcome == "failed" {
				foundFailed = true
			}
		}
	}
	if !foundSubmitted || !foundFailed {
		t.Fatalf("reports = %#v, want visible submitted and failed Turn reports", reports)
	}
}

func TestControllerProviderlessCanonicalTerminalCommitRetryConverges(t *testing.T) {
	t.Parallel()

	for _, test := range []struct {
		name                     string
		commitBeforeFirstFailure bool
	}{
		{name: "write failure"},
		{name: "commit success acknowledgment lost", commitBeforeFirstFailure: true},
	} {
		test := test
		t.Run(test.name, func(t *testing.T) {
			t.Parallel()

			adapter := &providerlessTerminalAcceptanceAdapter{
				recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
				failure: errors.New(
					"provider failed before durable acceptance",
				),
			}
			reporter := &retryingTerminalReporter{
				commitBeforeFirstFailure: test.commitBeforeFirstFailure,
			}
			controller := NewController([]Adapter{adapter}, reporter)
			sessionID := "session-providerless-terminal-commit-retry"
			turnID := "turn-providerless-terminal-commit-retry"
			if _, err := controller.Start(t.Context(), StartInput{
				RoomID: "room-1", AgentSessionID: sessionID,
				Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
			}); err != nil {
				t.Fatal(err)
			}

			result, err := controller.Exec(t.Context(), ExecInput{
				RoomID: "room-1", AgentSessionID: sessionID,
				TurnID: turnID, Content: textPrompt("hello"),
				RequireProviderAcceptance: true,
			})
			if err != nil {
				t.Fatalf("Exec() error = %v, want canonical submit retained", err)
			}
			if result.ProviderDispatch == nil ||
				result.ProviderDispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn {
				t.Fatalf(
					"provider dispatch = %#v, want applied_without_provider_turn",
					result.ProviderDispatch,
				)
			}
			waitForCondition(t, func() bool {
				return reporter.attempts() >= 1
			})
			if !controller.HasActiveTurn("room-1", sessionID) {
				t.Fatal("active-turn fence released before terminal retry committed")
			}
			waitForCondition(t, func() bool {
				return reporter.attempts() >= 2 &&
					!controller.HasActiveTurn("room-1", sessionID)
			})

			reports := reporter.snapshot()
			foundFailed := false
			for _, call := range reports {
				for _, patch := range call.report.StatePatches {
					if patch.Turn != nil && patch.Turn.TurnID == turnID &&
						patch.Turn.Phase == "settled" && patch.Turn.Outcome == "failed" {
						foundFailed = true
					}
				}
			}
			if !foundFailed {
				t.Fatalf("reports = %#v, want durable failed Turn", reports)
			}
		})
	}
}

// hangingProviderAcceptanceAdapter blocks inside ExecWithProviderAcceptance
// until the Exec context is canceled. It never reports an acceptance receipt,
// mirroring Claude cancel-before-identity-bound.
type hangingProviderAcceptanceAdapter struct {
	recordingStartAdapter
	entered chan struct{}
}

func (*hangingProviderAcceptanceAdapter) ForkCapabilities(
	context.Context,
	Session,
) (SessionForkCapabilities, error) {
	return SessionForkCapabilities{ThroughTurn: true}, nil
}

func (*hangingProviderAcceptanceAdapter) Fork(
	context.Context,
	SessionForkInput,
) (SessionForkResult, error) {
	return SessionForkResult{}, nil
}

func (a *hangingProviderAcceptanceAdapter) ExecWithProviderAcceptance(
	ctx context.Context,
	_ Session,
	_ []PromptContentBlock,
	_ string,
	_ string,
	_ EventSink,
	_ CommandSnapshotSink,
	_ ProviderDispatchSink,
	_ ProviderAcceptanceBarrier,
) ([]activityshared.Event, error) {
	if a.entered != nil {
		select {
		case a.entered <- struct{}{}:
		default:
		}
	}
	<-ctx.Done()
	return nil, ctx.Err()
}

func (*hangingProviderAcceptanceAdapter) UsesRootProviderTurnLifecycle() bool {
	return true
}

func TestControllerCancelBeforeProviderAcceptanceDoesNotLeaveDeliveryUnknown(t *testing.T) {
	t.Parallel()

	entered := make(chan struct{}, 1)
	adapter := &hangingProviderAcceptanceAdapter{
		recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
		entered:               entered,
	}
	controller := NewController([]Adapter{adapter}, &recordingReporter{})
	if _, err := controller.Start(t.Context(), StartInput{
		RoomID: "room-1", AgentSessionID: "session-cancel-before-accept",
		Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
	}); err != nil {
		t.Fatal(err)
	}

	execDone := make(chan struct {
		result ExecResult
		err    error
	}, 1)
	go func() {
		result, err := controller.Exec(t.Context(), ExecInput{
			RoomID: "room-1", AgentSessionID: "session-cancel-before-accept",
			TurnID: "turn-cancel-before-accept", Content: textPrompt("sleep 60"),
			ClientSubmitID:                  "submit-cancel-before-accept",
			CanonicalSubmitOccurredAtUnixMS: 1_700_000_000_001,
			RequireProviderAcceptance:       true,
		})
		execDone <- struct {
			result ExecResult
			err    error
		}{result: result, err: err}
	}()

	select {
	case <-entered:
	case outcome := <-execDone:
		t.Fatalf("Exec finished before provider acceptance: err=%v result=%#v", outcome.err, outcome.result)
	case <-t.Context().Done():
		t.Fatal("provider acceptance never entered")
	}
	waitForCondition(t, func() bool {
		return controller.HasActiveTurn("room-1", "session-cancel-before-accept")
	})
	if _, err := controller.Cancel(t.Context(), rootCancelInput(
		"room-1",
		"session-cancel-before-accept",
		"turn-cancel-before-accept",
		"user requested turn cancellation",
	)); err != nil {
		t.Fatalf("Cancel: %v", err)
	}

	outcome := <-execDone
	if outcome.err != nil {
		t.Fatalf("Exec() error = %v, want nil after cancel-before-acceptance", outcome.err)
	}
	if outcome.result.ProviderDispatch == nil ||
		outcome.result.ProviderDispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn ||
		outcome.result.ProviderDispatch.Acceptance != nil {
		t.Fatalf(
			"provider dispatch = %#v, want applied_without_provider_turn",
			outcome.result.ProviderDispatch,
		)
	}
	if outcome.result.TurnID != "turn-cancel-before-accept" {
		t.Fatalf("turn id = %q, want turn-cancel-before-accept", outcome.result.TurnID)
	}
	assertRootTurnFenceAwaitsCanonicalSettlement(
		t,
		controller,
		"session-cancel-before-accept",
		"turn-cancel-before-accept",
		"canceled",
	)
}

// preAcceptanceOutcomeUnknownAdapter reports outcome_unknown then returns
// without waiting for ctx cancel — matching Claude turn_canceled settling
// before Controller cancels runCtx.
type preAcceptanceOutcomeUnknownAdapter struct {
	recordingStartAdapter
	entered chan struct{}
}

func (*preAcceptanceOutcomeUnknownAdapter) ForkCapabilities(
	context.Context,
	Session,
) (SessionForkCapabilities, error) {
	return SessionForkCapabilities{ThroughTurn: true}, nil
}

func (*preAcceptanceOutcomeUnknownAdapter) Fork(
	context.Context,
	SessionForkInput,
) (SessionForkResult, error) {
	return SessionForkResult{}, nil
}

func (a *preAcceptanceOutcomeUnknownAdapter) ExecWithProviderAcceptance(
	_ context.Context,
	_ Session,
	_ []PromptContentBlock,
	_ string,
	_ string,
	_ EventSink,
	_ CommandSnapshotSink,
	reportDispatch ProviderDispatchSink,
	_ ProviderAcceptanceBarrier,
) ([]activityshared.Event, error) {
	if a.entered != nil {
		select {
		case a.entered <- struct{}{}:
		default:
		}
	}
	reportDispatch(ProviderDispatchResult{
		Disposition: DispatchDispositionOutcomeUnknown,
	})
	return nil, nil
}

func (*preAcceptanceOutcomeUnknownAdapter) UsesRootProviderTurnLifecycle() bool {
	return true
}

func TestControllerPreAcceptanceOutcomeUnknownDoesNotLeaveDeliveryUnknown(t *testing.T) {
	t.Parallel()

	adapter := &preAcceptanceOutcomeUnknownAdapter{
		recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
		entered:               make(chan struct{}, 1),
	}
	controller := NewController([]Adapter{adapter}, &recordingReporter{})
	if _, err := controller.Start(t.Context(), StartInput{
		RoomID: "room-1", AgentSessionID: "session-outcome-unknown",
		Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
	}); err != nil {
		t.Fatal(err)
	}

	result, err := controller.Exec(t.Context(), ExecInput{
		RoomID: "room-1", AgentSessionID: "session-outcome-unknown",
		TurnID: "turn-outcome-unknown", Content: textPrompt("sleep 60"),
		ClientSubmitID:                  "submit-outcome-unknown",
		CanonicalSubmitOccurredAtUnixMS: 1_700_000_000_002,
		RequireProviderAcceptance:       true,
	})
	if err != nil {
		t.Fatalf("Exec() error = %v, want nil", err)
	}
	if result.ProviderDispatch == nil ||
		result.ProviderDispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn ||
		result.ProviderDispatch.Acceptance != nil {
		t.Fatalf(
			"provider dispatch = %#v, want applied_without_provider_turn",
			result.ProviderDispatch,
		)
	}
	assertRootTurnFenceAwaitsCanonicalSettlement(
		t,
		controller,
		"session-outcome-unknown",
		"turn-outcome-unknown",
		"failed",
	)
}

func TestControllerCallerCancelDuringAcceptanceDoesNotLeaveDeliveryUnknown(t *testing.T) {
	t.Parallel()

	entered := make(chan struct{}, 1)
	adapter := &hangingProviderAcceptanceAdapter{
		recordingStartAdapter: recordingStartAdapter{provider: ProviderClaudeCode},
		entered:               entered,
	}
	controller := NewController([]Adapter{adapter}, &recordingReporter{})
	if _, err := controller.Start(t.Context(), StartInput{
		RoomID: "room-1", AgentSessionID: "session-caller-cancel",
		Provider: ProviderClaudeCode, CWD: "/workspace", Provisional: true,
	}); err != nil {
		t.Fatal(err)
	}

	callerCtx, callerCancel := context.WithCancel(t.Context())
	defer callerCancel()

	execDone := make(chan struct {
		result ExecResult
		err    error
	}, 1)
	go func() {
		result, err := controller.Exec(callerCtx, ExecInput{
			RoomID: "room-1", AgentSessionID: "session-caller-cancel",
			TurnID: "turn-caller-cancel", Content: textPrompt("sleep 60"),
			ClientSubmitID:                  "submit-caller-cancel",
			CanonicalSubmitOccurredAtUnixMS: 1_700_000_000_003,
			RequireProviderAcceptance:       true,
		})
		execDone <- struct {
			result ExecResult
			err    error
		}{result: result, err: err}
	}()

	select {
	case <-entered:
	case outcome := <-execDone:
		t.Fatalf("Exec finished before provider acceptance: err=%v result=%#v", outcome.err, outcome.result)
	case <-t.Context().Done():
		t.Fatal("provider acceptance never entered")
	}
	callerCancel()

	outcome := <-execDone
	if outcome.err != nil {
		t.Fatalf("Exec() error = %v, want nil after caller cancel", outcome.err)
	}
	if outcome.result.ProviderDispatch == nil ||
		outcome.result.ProviderDispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn ||
		outcome.result.ProviderDispatch.Acceptance != nil {
		t.Fatalf(
			"provider dispatch = %#v, want applied_without_provider_turn",
			outcome.result.ProviderDispatch,
		)
	}
	if !controller.HasActiveTurn("room-1", "session-caller-cancel") {
		t.Fatal("caller cancellation released root Turn before canonical settlement")
	}
	controller.cancelActiveTurn("room-1", "session-caller-cancel")
	assertRootTurnFenceAwaitsCanonicalSettlement(
		t,
		controller,
		"session-caller-cancel",
		"turn-caller-cancel",
		"canceled",
	)
}

func assertRootTurnFenceAwaitsCanonicalSettlement(
	t *testing.T,
	controller *Controller,
	sessionID string,
	turnID string,
	outcome string,
) {
	t.Helper()
	if !controller.HasActiveTurn("room-1", sessionID) {
		t.Fatalf("root Turn %q released before canonical settlement", turnID)
	}
	controller.ReconcileRootTurnSettlement(RootTurnSettlement{
		RoomID:         "room-1",
		AgentSessionID: sessionID,
		TurnID:         turnID,
		Outcome:        outcome,
	})
	waitForCondition(t, func() bool {
		return !controller.HasActiveTurn("room-1", sessionID)
	})
}