fix(session): settle late prompt completion as an intentional stop

lycaon/internal/session/submissions/admission.go

@@ -14,13 +14,11 @@ import (
 	"github.com/lycaon/lycaon/internal/noticeerr"
 	"github.com/lycaon/lycaon/internal/people"
 	"github.com/lycaon/lycaon/internal/promptresult"
-	"github.com/lycaon/lycaon/internal/session/lifecycle"
 	"github.com/lycaon/lycaon/internal/session/promptinput"
 	"github.com/lycaon/lycaon/internal/session/promptstate"
 	"github.com/lycaon/lycaon/internal/session/spendguard"
 	"github.com/lycaon/lycaon/internal/session/store"
 	"github.com/lycaon/lycaon/internal/session/submissionstate"
-	"github.com/lycaon/lycaon/pkg/api"
 )
 
 // AdmitPrompt persists a client-authored prompt before coordinator work begins.
@@ -409,33 +407,6 @@ func (m *Service) publishClosedWithoutTurn(ctx context.Context, row *store.Promp
 	}
 }
 
-// finishSubmission persists terminal state despite caller cancellation.
-func (m *Service) finishSubmission(ctx context.Context, row *store.PromptSubmission, resp *promptresult.Result, runErr error) error {
-	defer m.SubmissionState.Forget(row.ID)
-	status := store.PromptSubmissionComplete
-	var failure store.PromptSubmissionFailure
-	if runErr != nil {
-		status = store.PromptSubmissionFailed
-		failure.Message = runErr.Error()
-		// Preserve the structured code before serializing the error.
-		if code, ok := noticeerr.CodeOf(runErr); ok {
-			failure.Code = string(code)
-		}
-		if ctx.Err() != nil || errors.Is(runErr, lifecycle.ErrStopping) {
-			status = store.PromptSubmissionInterrupted
-		}
-	}
-	resultJSON := ""
-	if resp != nil {
-		encoded, marshalErr := json.Marshal(resp)
-		if marshalErr != nil {
-			return fmt.Errorf("encode prompt result: %w", marshalErr)
-		}
-		resultJSON = string(encoded)
-	}
-	return m.store.FinishPromptSubmission(context.WithoutCancel(ctx), row.ID, row.ClaimToken, status, resultJSON, failure)
-}
-
 // AbandonPromptSubmission fails only unclaimed receipts.
 func (m *Service) AbandonPromptSubmission(ctx context.Context, submissionID string, cause error) error {
 	if m == nil || m.store == nil {
@@ -473,33 +444,6 @@ func (m *Service) failUndecodableSubmission(ctx context.Context, submissionID st
 	return errors.Join(decodeErr, finishErr)
 }
 
-func Result(row *store.PromptSubmission) (*promptresult.Result, error) {
-	if row == nil {
-		return nil, nil
-	}
-	switch row.Status {
-	case store.PromptSubmissionRunning:
-		return nil, ErrInFlight
-	case store.PromptSubmissionCanceled:
-		return nil, fmt.Errorf("prompt submission %s was removed from the queue before it ran", row.ID)
-	case store.PromptSubmissionInterrupted:
-		// The sentinel lets receipt readers classify a stop like the runner did.
-		return nil, fmt.Errorf("prompt submission %s was interrupted: %s: %w", row.ID, row.Error, lifecycle.ErrStopping)
-	case store.PromptSubmissionFailed:
-		failed := fmt.Errorf("prompt submission %s is %s: %s", row.ID, row.Status, row.Error)
-		return nil, noticeerr.WithCode(failed, api.NoticeCode(row.ErrorCode))
-	case store.PromptSubmissionQueued, store.PromptSubmissionComplete:
-	}
-	if row.ResultJSON == "" {
-		return nil, nil
-	}
-	var resp promptresult.Result
-	if err := json.Unmarshal([]byte(row.ResultJSON), &resp); err != nil {
-		return nil, fmt.Errorf("decode prompt submission result: %w", err)
-	}
-	return &resp, nil
-}
-
 // RecoverPromptSubmissions resolves work interrupted by host exit.
 func (m *Service) RecoverPromptSubmissions(ctx context.Context) ([]string, error) {
 	if m == nil || m.store == nil {

lycaon/internal/session/submissions/completion.go

@@ -0,0 +1,78 @@
+package submissions
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/lycaon/lycaon/internal/noticeerr"
+	"github.com/lycaon/lycaon/internal/promptresult"
+	"github.com/lycaon/lycaon/internal/session/lifecycle"
+	"github.com/lycaon/lycaon/internal/session/store"
+	"github.com/lycaon/lycaon/pkg/api"
+)
+
+// finishSubmission persists terminal state despite caller cancellation.
+func (m *Service) finishSubmission(ctx context.Context, row *store.PromptSubmission, resp *promptresult.Result, runErr error) error {
+	defer m.SubmissionState.Forget(row.ID)
+	status := store.PromptSubmissionComplete
+	var failure store.PromptSubmissionFailure
+	if runErr != nil {
+		status = store.PromptSubmissionFailed
+		failure.Message = runErr.Error()
+		// Preserve the structured code before serializing the error.
+		if code, ok := noticeerr.CodeOf(runErr); ok {
+			failure.Code = string(code)
+		}
+		if ctx.Err() != nil || errors.Is(runErr, lifecycle.ErrStopping) {
+			status = store.PromptSubmissionInterrupted
+		}
+	}
+	resultJSON := ""
+	if resp != nil {
+		encoded, marshalErr := json.Marshal(resp)
+		if marshalErr != nil {
+			return fmt.Errorf("encode prompt result: %w", marshalErr)
+		}
+		resultJSON = string(encoded)
+	}
+	hostCtx := context.WithoutCancel(ctx)
+	err := m.store.FinishPromptSubmission(hostCtx, row.ID, row.ClaimToken, status, resultJSON, failure)
+	if err != nil {
+		// Stop may settle the same claim before the provider unwinds.
+		current, readErr := m.store.GetPromptSubmission(hostCtx, row.ID)
+		if readErr == nil && current != nil && row.ClaimToken != "" &&
+			current.ClaimToken == row.ClaimToken && current.Status == store.PromptSubmissionInterrupted {
+			return lifecycle.ErrStopping
+		}
+	}
+	return err
+}
+
+func Result(row *store.PromptSubmission) (*promptresult.Result, error) {
+	if row == nil {
+		return nil, nil
+	}
+	switch row.Status {
+	case store.PromptSubmissionRunning:
+		return nil, ErrInFlight
+	case store.PromptSubmissionCanceled:
+		return nil, fmt.Errorf("prompt submission %s was removed from the queue before it ran", row.ID)
+	case store.PromptSubmissionInterrupted:
+		// The sentinel lets receipt readers classify a stop like the runner did.
+		return nil, fmt.Errorf("prompt submission %s was interrupted: %s: %w", row.ID, row.Error, lifecycle.ErrStopping)
+	case store.PromptSubmissionFailed:
+		failed := fmt.Errorf("prompt submission %s is %s: %s", row.ID, row.Status, row.Error)
+		return nil, noticeerr.WithCode(failed, api.NoticeCode(row.ErrorCode))
+	case store.PromptSubmissionQueued, store.PromptSubmissionComplete:
+	}
+	if row.ResultJSON == "" {
+		return nil, nil
+	}
+	var resp promptresult.Result
+	if err := json.Unmarshal([]byte(row.ResultJSON), &resp); err != nil {
+		return nil, fmt.Errorf("decode prompt submission result: %w", err)
+	}
+	return &resp, nil
+}

lycaon/internal/session/submissions/completion_test.go

@@ -0,0 +1,101 @@
+package submissions
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/lycaon/lycaon/internal/session/lifecycle"
+	"github.com/lycaon/lycaon/internal/session/store"
+	"github.com/lycaon/lycaon/internal/session/submissionstate"
+	"github.com/lycaon/lycaon/internal/testdbfixture"
+	"github.com/lycaon/lycaon/internal/testdbseed"
+	"github.com/lycaon/lycaon/internal/testutil"
+	"github.com/lycaon/lycaon/pkg/api"
+)
+
+type completionStore interface {
+	Store
+	Create(context.Context, api.CreateSessionRequest, string) (*api.Session, error)
+	InterruptRunningPromptSubmissionsBySession(context.Context, string) error
+}
+
+func TestStoppedReceiptWinsLateCompletion(t *testing.T) {
+	for _, fixture := range []struct {
+		name string
+		open func(*testing.T) completionStore
+	}{
+		{"memory", func(*testing.T) completionStore { return store.NewMemory() }},
+		{"sql", func(t *testing.T) completionStore {
+			db := testdbfixture.Open(t, "completion.db")
+			testdbseed.InsertProjectRoot(t, db, testdbseed.DefaultProjectID, t.TempDir())
+			return store.NewSQL(db)
+		}},
+	} {
+		t.Run(fixture.name, func(t *testing.T) {
+			st := fixture.open(t)
+			sess, err := st.Create(t.Context(), api.CreateSessionRequest{}, testdbseed.DefaultProjectID)
+			testutil.FailErr(t, "create session", err)
+			for _, runErr := range []error{nil, context.Canceled, lifecycle.ErrStopping, errors.New("late provider failure")} {
+				row, _, err := st.PutPromptSubmission(t.Context(), store.PromptSubmission{
+					ID: uuid.NewString(), SessionID: sess.ID, ProjectID: sess.ProjectID,
+					InputDigest: "digest", InputJSON: `{}`, Origin: store.PromptSubmissionOriginUser, SubmittedBy: sess.OwnerPersonID,
+				})
+				testutil.FailErr(t, "admit prompt", err)
+				claimed, won, err := st.ClaimPromptSubmission(t.Context(), row.ID)
+				testutil.FailErr(t, "claim prompt", err)
+				if !won {
+					t.Fatal("prompt was not claimed")
+				}
+				testutil.FailErr(t, "stop session receipts", st.InterruptRunningPromptSubmissionsBySession(t.Context(), sess.ID))
+				before, err := st.GetPromptSubmission(t.Context(), row.ID)
+				testutil.FailErr(t, "read stopped receipt", err)
+				service := &Service{store: st, SubmissionState: &submissionstate.Service{}}
+				if err := service.finishSubmission(t.Context(), claimed, nil, runErr); !errors.Is(err, lifecycle.ErrStopping) {
+					t.Fatalf("late finish = %v, want a stop", err)
+				}
+				after, err := st.GetPromptSubmission(t.Context(), row.ID)
+				testutil.FailErr(t, "read settled receipt", err)
+				if !reflect.DeepEqual(before, after) {
+					t.Fatalf("stopped receipt changed: before=%+v after=%+v", before, after)
+				}
+				claimed.ClaimToken = "stale"
+				if err := service.finishSubmission(t.Context(), claimed, nil, nil); err == nil || errors.Is(err, lifecycle.ErrStopping) {
+					t.Fatalf("stale claim was mistaken for its own stop: %v", err)
+				}
+			}
+		})
+	}
+}
+
+type completionFailureStore struct {
+	Store
+	finishErr error
+	readErr   error
+	row       *store.PromptSubmission
+}
+
+func (s completionFailureStore) FinishPromptSubmission(context.Context, string, string, store.PromptSubmissionStatus, string, store.PromptSubmissionFailure) error {
+	return s.finishErr
+}
+func (s completionFailureStore) GetPromptSubmission(context.Context, string) (*store.PromptSubmission, error) {
+	return s.row, s.readErr
+}
+
+func TestCompletionPreservesUnconfirmedFailure(t *testing.T) {
+	failure := errors.New("write failed")
+	for _, current := range []*store.PromptSubmission{nil, {Status: store.PromptSubmissionRunning, ClaimToken: "claim"}, {Status: store.PromptSubmissionComplete, ClaimToken: "claim"}} {
+		st := completionFailureStore{finishErr: failure, row: current}
+		service := &Service{store: st, SubmissionState: &submissionstate.Service{}}
+		if err := service.finishSubmission(t.Context(), &store.PromptSubmission{ID: "id", ClaimToken: "claim"}, nil, nil); !errors.Is(err, failure) {
+			t.Fatalf("lost completion failure: %v", err)
+		}
+	}
+	st := completionFailureStore{finishErr: failure, readErr: errors.New("read failed"), row: &store.PromptSubmission{Status: store.PromptSubmissionInterrupted, ClaimToken: "claim"}}
+	service := &Service{store: st, SubmissionState: &submissionstate.Service{}}
+	if err := service.finishSubmission(t.Context(), &store.PromptSubmission{ID: "id", ClaimToken: "claim"}, nil, nil); !errors.Is(err, failure) {
+		t.Fatalf("unreadable receipt masked completion failure: %v", err)
+	}
+}

lycaon/internal/session/submissions/dispatch.go

@@ -12,6 +12,7 @@ import (
 	"github.com/lycaon/lycaon/internal/coordinator/loopwake"
 	"github.com/lycaon/lycaon/internal/coordinator/surface"
 	"github.com/lycaon/lycaon/internal/promptresult"
+	"github.com/lycaon/lycaon/internal/session/lifecycle"
 	"github.com/lycaon/lycaon/internal/session/promptinput"
 	"github.com/lycaon/lycaon/internal/session/store"
 	"github.com/lycaon/lycaon/internal/session/submissionstate"
@@ -244,8 +245,10 @@ func (m *Service) runQueuedTurn(ctx context.Context, id string, claimedRows []st
 	for _, row := range claimed {
 		if finishErr := m.finishSubmission(ctx, row, resp, err); finishErr != nil {
 			closeErr = errors.Join(closeErr, finishErr)
-			slog.ErrorContext(ctx, "queued prompt receipt close failed",
-				"session_id", id, "submission_id", row.ID, "err", finishErr)
+			if !errors.Is(finishErr, lifecycle.ErrStopping) {
+				slog.ErrorContext(ctx, "queued prompt receipt close failed",
+					"session_id", id, "submission_id", row.ID, "err", finishErr)
+			}
 		}
 	}
 	if !dispatch.Began() {

lycaon/internal/session/submissions/drain.go

@@ -2,7 +2,10 @@ package submissions
 
 import (
 	"context"
+	"errors"
 	"log/slog"
+
+	"github.com/lycaon/lycaon/internal/session/lifecycle"
 )
 
 // DrainQueue bypasses round-idle gating for a ready draft.
@@ -12,7 +15,7 @@ func (m *Service) DrainQueue(ctx context.Context, id string) {
 	}
 	ctx, unlockDispatch := m.LockDispatch(ctx, id)
 	defer unlockDispatch()
-	if _, err := m.dispatchPromptSubmissions(ctx, id, ""); err != nil {
+	if _, err := m.dispatchPromptSubmissions(ctx, id, ""); err != nil && !errors.Is(err, lifecycle.ErrStopping) {
 		slog.ErrorContext(ctx, "drain prompt submissions", "session_id", id, "error", err)
 	}
 }

lycaon/internal/session/turnadmission/queue_prompt.go

@@ -29,6 +29,9 @@ func (m *Service) runTurnAndDrain(ctx context.Context, id string, in promptinput
 		defer lock.Unlock()
 		return m.Runner.Run(ctx, id, in)
 	}()
+	if !m.Gate.MayDrain(turn) && onlyCanceled(err) {
+		err = lifecycle.ErrStopping
+	}
 	LogTurnFailure(ctx, id, err)
 	err = m.Turns.ReportFailure(ctx, id, err)
 	if m.Gate.MayDrain(turn) {
@@ -46,6 +49,22 @@ func (m *Service) runTurnAndDrain(ctx context.Context, id string, in promptinput
 	return resp, err
 }
 
+func onlyCanceled(err error) bool {
+	if joined, ok := err.(interface{ Unwrap() []error }); ok {
+		causes := joined.Unwrap()
+		for _, cause := range causes {
+			if !onlyCanceled(cause) {
+				return false
+			}
+		}
+		return len(causes) != 0
+	}
+	if wrapped, ok := err.(interface{ Unwrap() error }); ok {
+		return onlyCanceled(wrapped.Unwrap())
+	}
+	return errors.Is(err, context.Canceled)
+}
+
 func LogTurnFailure(ctx context.Context, sessionID string, err error) {
 	if err == nil || errors.Is(err, lifecycle.ErrStopping) {
 		return