fix(resume): preserve checkpoints after Ctrl-C

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.

cmd/opencodereview/review_cmd.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"os/signal"
 	"path/filepath"
 	"sort"
 	"strings"
@@ -96,15 +97,17 @@ 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)
 	},
 }
 
 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,

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")

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 {

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()

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

pages/src/content/docs/ja/cli-reference.md

@@ -182,8 +182,10 @@ ocr review --commit abc123 --resume <session-id>
 - provider や model の変更は `--provider` / `--model` で明示的に指定する必要が
   あります。設定ファイルや環境変数経由の変更は拒否されます
 - 親の実行が run manifest を持っている必要があります。入力はこれと照合して
-  検証されます。Ctrl-C で中断された実行は書き出しておらず、run manifest より
-  古いセッションはそもそも持っていません
+  検証されます。ファイルの dispatch 開始後は、Ctrl-C によってレビューが正常に
+  キャンセルされて manifest が書き出されるため、完了済みの checkpoint は再開時に
+  再利用できます。正常に終了できなかったプロセスと run manifest より古い
+  セッションには manifest がありません
 - 再利用されるのは、親の manifest が結果を確定したファイルだけです。manifest が
   裏付けないチェックポイントや読み取れないチェックポイントは、そのファイルが
   もう一度レビューされるだけで、他のファイルには影響しません

pages/src/content/docs/ru/cli-reference.md

@@ -178,8 +178,10 @@ ocr review --commit abc123 --resume <session-id>
 - смену provider или model нужно запросить явно через `--provider` / `--model`;
   изменение, пришедшее из конфигурации или окружения, отклоняется;
 - у родительского запуска должен быть run manifest — именно по нему проверяется
-  вход. Запуск, прерванный Ctrl-C, его не записал, а сессии старше run manifest
-  его никогда и не имели;
+  вход. После начала dispatch файлов Ctrl-C корректно отменяет ревью и записывает
+  manifest, поэтому завершённые checkpoint можно использовать при возобновлении.
+  Процесс, завершённый без корректного закрытия, и сессии старше run manifest не
+  имеют manifest;
 - переиспользуются только файлы, судьбу которых зафиксировал manifest родителя.
   Контрольная точка, которую manifest не подтверждает или которую не удалось
   прочитать, стоит этому файлу его контрольной точки и не более — он просто

pages/src/content/docs/zh/cli-reference.md

@@ -180,8 +180,9 @@ ocr review --commit abc123 --resume <session-id>
   改变了选中的文件集合，整次恢复会被拒绝，而不是部分复用
 - 切换 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` 不能同时使用