diff --git a/lycaon/internal/session/submissions/admission.go b/lycaon/internal/session/submissions/admission.go index aa384c36e1..409d1c7339 100644 --- a/lycaon/internal/session/submissions/admission.go +++ b/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 { diff --git a/lycaon/internal/session/submissions/completion.go b/lycaon/internal/session/submissions/completion.go new file mode 100644 index 0000000000..d497d94fd9 --- /dev/null +++ b/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 +} diff --git a/lycaon/internal/session/submissions/completion_test.go b/lycaon/internal/session/submissions/completion_test.go new file mode 100644 index 0000000000..c3304123f9 --- /dev/null +++ b/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) + } +} diff --git a/lycaon/internal/session/submissions/dispatch.go b/lycaon/internal/session/submissions/dispatch.go index 1823ce612b..a95ce9b382 100644 --- a/lycaon/internal/session/submissions/dispatch.go +++ b/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() { diff --git a/lycaon/internal/session/submissions/drain.go b/lycaon/internal/session/submissions/drain.go index 6d067f7b16..99aaad5f13 100644 --- a/lycaon/internal/session/submissions/drain.go +++ b/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) } } diff --git a/lycaon/internal/session/turnadmission/queue_prompt.go b/lycaon/internal/session/turnadmission/queue_prompt.go index ca6f886e4a..1ddc3872f0 100644 --- a/lycaon/internal/session/turnadmission/queue_prompt.go +++ b/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 diff --git a/lycaon/internal/session/turnadmission/queue_prompt_test.go b/lycaon/internal/session/turnadmission/queue_prompt_test.go new file mode 100644 index 0000000000..eef91a6b26 --- /dev/null +++ b/lycaon/internal/session/turnadmission/queue_prompt_test.go @@ -0,0 +1,100 @@ +package turnadmission + +import ( + "context" + "errors" + "fmt" + "testing" + + "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" +) + +func TestRunLockedClassifiesCancellationBeforeReporting(t *testing.T) { + cleanup := errors.New("cleanup failed") + for _, tc := range []struct { + name string + stop string + runErr error + wantStop bool + }{ + {name: "active stop", stop: "active", runErr: context.Canceled, wantStop: true}, + {name: "finished stop", stop: "finished", runErr: context.Canceled, wantStop: true}, + {name: "wrapped cancellation", stop: "finished", runErr: fmt.Errorf("provider: %w", context.Canceled), wantStop: true}, + {name: "joined cancellations", stop: "active", runErr: errors.Join(context.Canceled, context.Canceled), wantStop: true}, + {name: "unexpected cancellation", runErr: context.Canceled}, + {name: "another session stopped", stop: "other", runErr: context.Canceled}, + {name: "joined cleanup failure", stop: "active", runErr: errors.Join(context.Canceled, cleanup)}, + {name: "wrapped joined cleanup", stop: "finished", runErr: fmt.Errorf("provider: %w", errors.Join(context.Canceled, cleanup))}, + {name: "deadline", stop: "active", runErr: context.DeadlineExceeded}, + {name: "provider failure", stop: "active", runErr: cleanup}, + {name: "completed result", stop: "finished"}, + } { + t.Run(tc.name, func(t *testing.T) { + gate := lifecycle.New(nil) + turns := &recordingTurns{} + settlement := &recordingSettlement{} + service := &Service{Gate: gate, Workspace: readyWorkspace{}, Turns: turns, Settlement: settlement} + service.Runner = promptRunner(func() (*promptresult.Result, error) { + if tc.stop != "" { + id := "session" + if tc.stop == "other" { + id = "other" + } + flight, _ := gate.Begin(id) + if tc.stop == "finished" { + gate.Finish(id, flight, nil) + } + } + return nil, tc.runErr + }) + var locks promptstate.MutexRegistry + lock := locks.Acquire("session") + lock.Lock() + _, got := service.RunLocked(t.Context(), "session", promptinput.Input{}, lock) + if tc.wantStop { + if !errors.Is(got, lifecycle.ErrStopping) || !errors.Is(turns.reported, lifecycle.ErrStopping) { + t.Fatalf("stop classification: returned %v, reported %v", got, turns.reported) + } + } else if got != tc.runErr || turns.reported != tc.runErr { + t.Fatalf("failure changed: returned %v, reported %v, want %v", got, turns.reported, tc.runErr) + } + wantDrain := tc.stop == "" || tc.stop == "other" + if settlement.drained != wantDrain { + t.Fatalf("drained = %v, want %v", settlement.drained, wantDrain) + } + probe := locks.Acquire("session") + if !probe.TryLock() { + t.Fatal("prompt lock was not released") + } + probe.Unlock() + }) + } +} + +type promptRunner func() (*promptresult.Result, error) + +func (f promptRunner) Run(context.Context, string, promptinput.Input) (*promptresult.Result, error) { + return f() +} + +type readyWorkspace struct{} + +func (readyWorkspace) CheckReady(context.Context, string) error { return nil } + +type recordingTurns struct{ reported error } + +func (r *recordingTurns) ReportFailure(_ context.Context, _ string, err error) error { + r.reported = err + return err +} +func (*recordingTurns) HostTurnBlocked(context.Context, string) bool { return false } + +type recordingSettlement struct{ drained bool } + +func (r *recordingSettlement) Drain(context.Context, string) error { + r.drained = true + return nil +}