diff --git a/cmd/opencodereview/compat_test.go b/cmd/opencodereview/compat_test.go index 462a8ca5..c14a17af 100644 --- a/cmd/opencodereview/compat_test.go +++ b/cmd/opencodereview/compat_test.go @@ -3,7 +3,15 @@ package main -import "github.com/spf13/cobra" +import ( + "context" + + "github.com/spf13/cobra" +) + +func executeReview(opts reviewOptions) error { + return executeReviewContext(context.Background(), opts) +} // parseReviewFlags provides test compatibility: parses args through a fresh // cobra command instance and returns the resulting reviewOptions. diff --git a/cmd/opencodereview/review_cmd.go b/cmd/opencodereview/review_cmd.go index 77aa7a29..1d1e8dd2 100644 --- a/cmd/opencodereview/review_cmd.go +++ b/cmd/opencodereview/review_cmd.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "os" + "os/signal" "path/filepath" "sort" "strings" @@ -96,7 +97,9 @@ var reviewCmd = &cobra.Command{ if err := validateReviewOptions(&reviewOpts); err != nil { return err } - return executeReview(reviewOpts) + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) + defer stop() + return executeReviewContext(ctx, reviewOpts) }, } @@ -104,7 +107,7 @@ func init() { registerReviewFlags(reviewCmd, &reviewOpts) } -func executeReview(opts reviewOptions) error { +func executeReviewContext(ctx context.Context, opts reviewOptions) error { cc, err := loadCommonContext(opts.repoDir, opts.rulePath, opts.maxTools, opts.maxGitProcs, true) if err != nil { return err @@ -137,7 +140,7 @@ func executeReview(opts reviewOptions) error { } if opts.preview { - return runPreview(cc, opts) + return runPreviewContext(ctx, cc, opts) } resumeState, err := loadReviewResumeState(cc.RepoDir, opts) @@ -162,7 +165,7 @@ func executeReview(opts reviewOptions) error { // Strictly before agent.New, so a rejected resume persists nothing. The sealed // input it returns pins the run to the very commits this check passed on, so // the decision cannot be undone by a ref moving afterwards. - sealed, err := validateResumeIdentity(context.Background(), cc, opts, rt, resumeState) + sealed, err := validateResumeIdentity(ctx, cc, opts, rt, resumeState) if err != nil { return err } @@ -186,7 +189,7 @@ func executeReview(opts reviewOptions) error { } tools := buildToolRegistry(rt.Collector, fileReader) - mcpClients := initMCPClients(context.Background(), rt.AppCfg, tools, cc.RepoDir, Version) + mcpClients := initMCPClients(ctx, rt.AppCfg, tools, cc.RepoDir, Version) defer func() { for _, mc := range mcpClients { if err := mc.Close(); err != nil { @@ -232,7 +235,7 @@ func executeReview(opts reviewOptions) error { q := newQuietHandle(opts.outputFormat, opts.audience) defer q.Restore() - ctx, span := telemetry.StartSpan(telemetry.ContextWithTraceParentFromEnv(context.Background()), "review.run") + runCtx, span := telemetry.StartSpan(telemetry.ContextWithTraceParentFromEnv(ctx), "review.run") defer span.End() telemetry.SetAttr(span, "review.repo", cc.RepoDir) telemetry.SetAttr(span, "review.from", opts.from) @@ -247,7 +250,7 @@ func executeReview(opts reviewOptions) error { } startTime := time.Now() - comments, runErr := ag.Run(ctx) + comments, runErr := ag.Run(runCtx) manifest := ag.RunManifest() // Freeze the retry report at the same boundary as the manifest: ag.Run has @@ -277,7 +280,7 @@ func executeReview(opts reviewOptions) error { var emitErr error emitted := manifest != nil || runErr == nil if emitted { - emitErr = emitRunResult(ctx, ag, comments, startTime, opts.outputFormat, opts.audience, q, llmIdentity, retryReport) + emitErr = emitRunResult(runCtx, ag, comments, startTime, opts.outputFormat, opts.audience, q, llmIdentity, retryReport) if emitErr != nil { emitErr = fmt.Errorf("emit review result: %w", emitErr) } @@ -478,8 +481,8 @@ func validateReviewRefs(repoDir string, opts reviewOptions) error { return nil } -func runPreview(cc *commonContext, opts reviewOptions) error { - preview, err := agent.Preview(context.Background(), agent.Args{ +func runPreviewContext(ctx context.Context, cc *commonContext, opts reviewOptions) error { + preview, err := agent.Preview(ctx, agent.Args{ RepoDir: cc.RepoDir, From: opts.from, To: opts.to, diff --git a/cmd/opencodereview/review_helpers_test.go b/cmd/opencodereview/review_helpers_test.go index 4364f911..c9feac55 100644 --- a/cmd/opencodereview/review_helpers_test.go +++ b/cmd/opencodereview/review_helpers_test.go @@ -11,6 +11,10 @@ import ( "github.com/alibaba/open-code-review/internal/tool" ) +func runPreview(cc *commonContext, opts reviewOptions) error { + return runPreviewContext(context.Background(), cc, opts) +} + func TestRunPreview(t *testing.T) { dir := initTestGitRepo(t) gitCommitFile(t, dir, "x.go", "package x\n", "add x") diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 40e1c506..dd8e24d3 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -606,6 +606,7 @@ func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error timeout := time.Duration(a.args.ConcurrentTaskTimeout) * time.Minute var dispatched int64 +dispatchLoop: for i := range toDispatch { if toDispatch[i].IsDeleted { continue @@ -655,9 +656,17 @@ func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error } } + select { + case sem <- struct{}{}: // acquire semaphore + case <-ctx.Done(): + break dispatchLoop + } + if ctx.Err() != nil { + <-sem // release the slot acquired concurrently with cancellation + break dispatchLoop + } dispatched++ wg.Add(1) - sem <- struct{}{} // acquire semaphore go func(d model.Diff) { fingerprint := reviewItemFingerprint(a.reviewMode(), d) @@ -731,15 +740,18 @@ func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error } wg.Wait() - - if dispatched == 0 { - return a.args.CommentCollector.Comments(), nil - } - // All subtasks finished — collect comments from the global collector once. if a.args.CommentWorkerPool != nil { a.args.CommentWorkerPool.Await() } + if ctxErr := ctx.Err(); ctxErr != nil { + a.recordContextFailure(ctxErr) + return a.args.CommentCollector.Comments(), ctxErr + } + + if dispatched == 0 { + return a.args.CommentCollector.Comments(), nil + } failed := atomic.LoadInt64(&a.subtaskFailed) reused := int64(0) @@ -756,6 +768,22 @@ func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error return a.args.CommentCollector.Comments(), nil } +func (a *Agent) recordContextFailure(err error) { + if b := a.session.Manifest(); b != nil { + var setErr error + if errors.Is(err, context.DeadlineExceeded) { + // A deadline truncates pending coverage without overriding completed items. + setErr = b.SetPendingFailureCause(session.FailureTimeout, "review deadline exceeded") + } else { + // Explicit cancellation stops the run itself, not just its pending items. + setErr = b.SetRunFailure(session.RunFailureCancelled, "review was cancelled") + } + if setErr != nil { + a.recordWarning("manifest_error", "", setErr.Error()) + } + } +} + func (a *Agent) applyResume(diffs []model.Diff) []model.Diff { resume := a.args.Resume if resume == nil { diff --git a/internal/agent/manifest_integration_test.go b/internal/agent/manifest_integration_test.go index 5a823e35..a1ca52d9 100644 --- a/internal/agent/manifest_integration_test.go +++ b/internal/agent/manifest_integration_test.go @@ -9,6 +9,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "testing" "time" @@ -42,6 +43,26 @@ func (manifestFlowClient) CompletionsWithCtx(_ context.Context, req llm.ChatRequ } } +type cancellationFlowClient struct { + blocked chan struct{} + once sync.Once +} + +func (c *cancellationFlowClient) CompletionsWithCtx(ctx context.Context, req llm.ChatRequest) (*llm.ChatResponse, error) { + var prompt string + for _, message := range req.Messages { + if text, ok := message.Content.(string); ok { + prompt += text + } + } + if strings.Contains(prompt, "blocked.go") { + c.once.Do(func() { close(c.blocked) }) + <-ctx.Done() + return nil, ctx.Err() + } + return agentTaskDoneResponse(), nil +} + func newManifestFlowAgent(t *testing.T, diffs []model.Diff, resume *session.ResumeState) *Agent { t.Helper() return newManifestFlowAgentWithClient(t, diffs, resume, manifestFlowClient{}) @@ -168,6 +189,101 @@ func TestManifestFlowCompleteAndPartial(t *testing.T) { }) } +func TestManifestFlowCancellationPersistsResumableSession(t *testing.T) { + done := model.Diff{OldPath: "done.go", NewPath: "done.go", Diff: "+done", Insertions: 1} + blocked := model.Diff{OldPath: "blocked.go", NewPath: "blocked.go", Diff: "+blocked", Insertions: 1} + pending := model.Diff{OldPath: "pending.go", NewPath: "pending.go", Diff: "+pending", Insertions: 1} + client := &cancellationFlowClient{blocked: make(chan struct{})} + a := newManifestFlowAgentWithClient(t, []model.Diff{done, blocked, pending}, nil, client) + a.args.MaxConcurrency = 1 + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchErr := make(chan error, 1) + go func() { + _, err := a.dispatchSubtasks(ctx) + dispatchErr <- err + }() + + select { + case <-client.blocked: + cancel() + case <-time.After(5 * time.Second): + t.Fatal("blocked review did not start") + } + if err := <-dispatchErr; !errors.Is(err, context.Canceled) { + t.Fatalf("dispatch error = %v, want context.Canceled", err) + } + + manifest := finishManifestFlow(t, a) + if manifest.RunFailure == nil || manifest.RunFailure.Classification != session.RunFailureCancelled { + t.Fatalf("run failure = %+v, want cancelled", manifest.RunFailure) + } + if len(manifest.Coverage.Completed) != 1 || len(manifest.Coverage.Failed) != 2 { + t.Fatalf("coverage = %+v, want one completed and two failed", manifest.Coverage) + } + for _, item := range manifest.Coverage.Failed { + if item.Classification != session.FailureCancelled { + t.Fatalf("failed item = %+v, want cancelled", item) + } + } + + state, err := session.LoadReviewResumeState(a.args.RepoDir, a.session.SessionID) + if err != nil { + t.Fatalf("load cancelled session: %v", err) + } + identity := session.RunIdentity{ + Mode: manifest.Input.Mode, + SourceArtifactSHA256: manifest.Input.SourceArtifactSHA256, + RuleConfigSHA256: manifest.Execution.RuleConfigSHA256, + RepositorySHA256: manifest.Repository.IdentitySHA256, + } + if err := state.ValidateResume(session.ResumeRequest{ + Identity: identity, + Provider: manifest.Execution.Provider, + Model: manifest.Execution.Model, + }); err != nil { + t.Fatalf("cancelled session should remain resumable: %v", err) + } + if _, ok := state.ReusableItem(reviewItemFingerprint(a.reviewMode(), done)); !ok { + t.Fatal("completed item is not reusable after cancellation") + } + if _, ok := state.ReusableItem(reviewItemFingerprint(a.reviewMode(), blocked)); ok { + t.Fatal("cancelled item must be reviewed again") + } +} + +func TestManifestFlowCancellationBeforeDispatchStartsNoSubtask(t *testing.T) { + pending := model.Diff{OldPath: "blocked.go", NewPath: "blocked.go", Diff: "+blocked", Insertions: 1} + client := &cancellationFlowClient{blocked: make(chan struct{})} + a := newManifestFlowAgentWithClient(t, []model.Diff{pending}, nil, client) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := a.dispatchSubtasks(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("dispatch error = %v, want context.Canceled", err) + } + if got := atomic.LoadInt64(&a.subtaskFailed); got != 0 { + t.Fatalf("subtask failures = %d, want 0 because no subtask should start", got) + } + select { + case <-client.blocked: + t.Fatal("LLM client was called after cancellation") + default: + } + + manifest := finishManifestFlow(t, a) + if manifest.RunFailure == nil || manifest.RunFailure.Classification != session.RunFailureCancelled { + t.Fatalf("run failure = %+v, want cancelled", manifest.RunFailure) + } + if len(manifest.Coverage.Completed) != 0 || len(manifest.Coverage.Failed) != 1 { + t.Fatalf("coverage = %+v, want one cancelled item and no completed items", manifest.Coverage) + } + if got := manifest.Coverage.Failed[0].Classification; got != session.FailureCancelled { + t.Fatalf("failure classification = %q, want %q", got, session.FailureCancelled) + } +} + func TestManifestFlowRunInputFailureIsPersisted(t *testing.T) { t.Setenv("HOME", t.TempDir()) repoDir := t.TempDir() diff --git a/pages/src/content/docs/en/cli-reference.md b/pages/src/content/docs/en/cli-reference.md index 5f587c68..8016216d 100644 --- a/pages/src/content/docs/en/cli-reference.md +++ b/pages/src/content/docs/en/cli-reference.md @@ -195,8 +195,9 @@ would review the same thing the parent did: - a provider or model change must be asked for explicitly with `--provider` / `--model`. A change that arrived through config or the environment is rejected - the parent must carry a run manifest, which is what its input is verified - against. A run killed with Ctrl-C never wrote one, and sessions older than run - manifests never had one + against. After file dispatch begins, Ctrl-C cancels the review gracefully and + records one, so completed checkpoints remain resumable. A process killed + before graceful shutdown and sessions older than run manifests do not have one - only files the parent's manifest settled are reused. A checkpoint the manifest does not account for, or one that is unreadable, costs that file its checkpoint and nothing more — it is simply reviewed again diff --git a/pages/src/content/docs/ja/cli-reference.md b/pages/src/content/docs/ja/cli-reference.md index 47a13a12..4c02b295 100644 --- a/pages/src/content/docs/ja/cli-reference.md +++ b/pages/src/content/docs/ja/cli-reference.md @@ -182,8 +182,10 @@ ocr review --commit abc123 --resume - provider や model の変更は `--provider` / `--model` で明示的に指定する必要が あります。設定ファイルや環境変数経由の変更は拒否されます - 親の実行が run manifest を持っている必要があります。入力はこれと照合して - 検証されます。Ctrl-C で中断された実行は書き出しておらず、run manifest より - 古いセッションはそもそも持っていません + 検証されます。ファイルの dispatch 開始後は、Ctrl-C によってレビューが正常に + キャンセルされて manifest が書き出されるため、完了済みの checkpoint は再開時に + 再利用できます。正常に終了できなかったプロセスと run manifest より古い + セッションには manifest がありません - 再利用されるのは、親の manifest が結果を確定したファイルだけです。manifest が 裏付けないチェックポイントや読み取れないチェックポイントは、そのファイルが もう一度レビューされるだけで、他のファイルには影響しません diff --git a/pages/src/content/docs/ru/cli-reference.md b/pages/src/content/docs/ru/cli-reference.md index af95256a..4a9be2d2 100644 --- a/pages/src/content/docs/ru/cli-reference.md +++ b/pages/src/content/docs/ru/cli-reference.md @@ -178,8 +178,10 @@ ocr review --commit abc123 --resume - смену provider или model нужно запросить явно через `--provider` / `--model`; изменение, пришедшее из конфигурации или окружения, отклоняется; - у родительского запуска должен быть run manifest — именно по нему проверяется - вход. Запуск, прерванный Ctrl-C, его не записал, а сессии старше run manifest - его никогда и не имели; + вход. После начала dispatch файлов Ctrl-C корректно отменяет ревью и записывает + manifest, поэтому завершённые checkpoint можно использовать при возобновлении. + Процесс, завершённый без корректного закрытия, и сессии старше run manifest не + имеют manifest; - переиспользуются только файлы, судьбу которых зафиксировал manifest родителя. Контрольная точка, которую manifest не подтверждает или которую не удалось прочитать, стоит этому файлу его контрольной точки и не более — он просто diff --git a/pages/src/content/docs/zh/cli-reference.md b/pages/src/content/docs/zh/cli-reference.md index 7f6bc3ff..ac268338 100644 --- a/pages/src/content/docs/zh/cli-reference.md +++ b/pages/src/content/docs/zh/cli-reference.md @@ -180,8 +180,9 @@ ocr review --commit abc123 --resume 改变了选中的文件集合，整次恢复会被拒绝，而不是部分复用 - 切换 provider 或 model 必须通过 `--provider` / `--model` 显式声明；经由配置 文件或环境变量发生的变化一律拒绝 -- 父运行必须带有 run manifest，输入正是拿它来校验的。被 Ctrl-C 终止的运行没写出 - manifest，早于 run manifest 的老 session 则从来就没有 +- 父运行必须带有 run manifest，输入正是拿它来校验的。文件派发开始后，Ctrl-C 会 + 优雅取消评审并写出 manifest，因此已完成的 checkpoint 仍可恢复；未能优雅关闭的 + 进程和早于 run manifest 的老 session 则没有 manifest - 只有父 manifest 认领过的文件才会复用。manifest 未认领或已损坏的 checkpoint 只 影响它自己那个文件——该文件重新评审一次，其余不受影响 - `--preview` 和 `--resume` 不能同时使用