mirror of
https://github.com/alibaba/open-code-review.git
synced 2026-10-02 09:24:52 +08:00
refactor: retire per-file naming left over from grouped review (#1168)
* refactor(llmloop): rename RunPerFile to RunMainTask The loop stopped being per-file when review switched from one subtask per file to one subtask per group: internal/agent passes a groupKey that may name several files, while internal/scan still passes a single path. The name asserted a granularity only one of the two callers has. RunMainTask names what the function actually drives — one MAIN_TASK conversation — matching the vocabulary already in place around it (MainLoopStop, the main.loop span, session.MainTask). RunSubtask would have been a second misfit, since review calls this once per review round and one subtask spans several rounds. The newPath parameter is renamed taskKey along the whole private chain (runGraceRound, executeToolCall, addNextMessage, runCompression, triggerAsyncCompression, requestCtx). Deps.NewRequestMeta keeps filePath: the session layer still spells that slot "file path" in its persisted records, and requestCtx now documents the seam instead of hiding it. ToolFailureDetail.FilePath is where a group key becomes externally visible — it reaches --format json as file_path — so the field now says so rather than letting a consumer assume it parses as one path. Two comments claimed executeToolCall overrides a comment's path with the loop's path argument. It has always been a fallback that applies only when the model named no path, and the difference decides whether a grouped review files its comments under the right files, so both now describe the fallback. Pure rename plus comment edits; no behavior change. * refactor(cli): retire per-file naming left over from grouped review perFileTimeout was the last holdout of the old granularity in the CLI options: its flag is --timeout, its help already said "concurrent task timeout", and it feeds Args.ConcurrentTaskTimeout. Only the field name still claimed a per-file deadline it never set. Two review help strings also outlived the change to grouped subtasks, and disagreed with cli-reference.md, which already documents both as per-group: --concurrency dispatches file groups, and --max-tokens caps a group's prompt. The scan command keeps its per-file wording, where one subtask is still one file. Same for six agent comments that describe per-group work as per-file — the budget look-ahead, the dispatch step and its function doc, the only deadline classifyItemError can see, and the goroutines allDiffs is read from — plus Run's pipeline summary, which omitted the grouping step altogether. Granularity that really is per-file stays as it is, and the two are easy to confuse: filterLargeDiffs measures each file's diff against the token ceiling on its own, review items and their status badges are per file, and a group's prompt embeds one XML element per file. The comment in review_cmd.go now names filterLargeDiffs so the distinction is checkable rather than a matter of trusting the adjective. The estimator keeps both its name avgMainRoundsPerFile and its arithmetic: it still bills every file for its own MAIN_TASK rounds, so a group of files is over-estimated. That is a costing bug rather than a naming one and needs its own change; only estimateDiffFileTokens' doc comment is corrected here, since the look-ahead consuming it sums per group.
This commit is contained in:
@@ -29,30 +29,30 @@ import (
|
||||
)
|
||||
|
||||
type reviewOptions struct {
|
||||
toolConfigPath string
|
||||
rulePath string
|
||||
repoDir string
|
||||
from string
|
||||
to string
|
||||
commit string
|
||||
resume string
|
||||
excludes string
|
||||
outputFormat string
|
||||
audience string
|
||||
outputPath string
|
||||
background string
|
||||
backgroundFile string
|
||||
provider string
|
||||
model string
|
||||
concurrency int
|
||||
perFileTimeout int
|
||||
maxTools int
|
||||
maxGitProcs int
|
||||
maxTokens int
|
||||
maxTokensBudget int
|
||||
effort string
|
||||
noFilter bool
|
||||
preview bool
|
||||
toolConfigPath string
|
||||
rulePath string
|
||||
repoDir string
|
||||
from string
|
||||
to string
|
||||
commit string
|
||||
resume string
|
||||
excludes string
|
||||
outputFormat string
|
||||
audience string
|
||||
outputPath string
|
||||
background string
|
||||
backgroundFile string
|
||||
provider string
|
||||
model string
|
||||
concurrency int
|
||||
concurrentTaskTimeout int
|
||||
maxTools int
|
||||
maxGitProcs int
|
||||
maxTokens int
|
||||
maxTokensBudget int
|
||||
effort string
|
||||
noFilter bool
|
||||
preview bool
|
||||
}
|
||||
|
||||
var reviewOpts reviewOptions
|
||||
@@ -223,7 +223,7 @@ func executeReviewContext(ctx context.Context, opts reviewOptions) (retErr error
|
||||
CommentCollector: rt.Collector,
|
||||
CommentWorkerPool: agent.NewCommentWorkerPool(opts.concurrency),
|
||||
MaxConcurrency: opts.concurrency,
|
||||
ConcurrentTaskTimeout: opts.perFileTimeout,
|
||||
ConcurrentTaskTimeout: opts.concurrentTaskTimeout,
|
||||
Model: rt.Model,
|
||||
Provider: rt.Provider,
|
||||
Background: opts.background,
|
||||
@@ -373,8 +373,9 @@ func loadReviewResumeState(repoDir string, opts reviewOptions) (*session.ResumeS
|
||||
// It must run before agent.New: agent.New creates the session, and session.New
|
||||
// writes session_start immediately, so validating any later would leave an orphan
|
||||
// session on disk behind every rejection. It must also run after max-tokens is
|
||||
// resolved, because the per-file token ceiling decides which large diffs are
|
||||
// dropped and therefore which files the input identity covers.
|
||||
// resolved, because agent.filterLargeDiffs measures each file's diff against
|
||||
// that ceiling on its own — grouping never enters this decision — and what it
|
||||
// drops is what the input identity stops covering.
|
||||
//
|
||||
// provider and model are explicit exactly when their flag was passed on this
|
||||
// command line: both default to the empty string and nothing else can set them,
|
||||
|
||||
@@ -25,29 +25,29 @@ import (
|
||||
)
|
||||
|
||||
type scanOptions struct {
|
||||
toolConfigPath string
|
||||
rulePath string
|
||||
repoDir string
|
||||
paths string
|
||||
excludes string
|
||||
outputFormat string
|
||||
audience string
|
||||
outputPath string
|
||||
background string
|
||||
concurrency int
|
||||
perFileTimeout int
|
||||
maxTools int
|
||||
maxGitProcs int
|
||||
preview bool
|
||||
noPlan bool
|
||||
noDedup bool
|
||||
noSummary bool
|
||||
batch string
|
||||
maxTokens int
|
||||
maxTokensBudget int
|
||||
provider string
|
||||
model string
|
||||
resume string
|
||||
toolConfigPath string
|
||||
rulePath string
|
||||
repoDir string
|
||||
paths string
|
||||
excludes string
|
||||
outputFormat string
|
||||
audience string
|
||||
outputPath string
|
||||
background string
|
||||
concurrency int
|
||||
concurrentTaskTimeout int
|
||||
maxTools int
|
||||
maxGitProcs int
|
||||
preview bool
|
||||
noPlan bool
|
||||
noDedup bool
|
||||
noSummary bool
|
||||
batch string
|
||||
maxTokens int
|
||||
maxTokensBudget int
|
||||
provider string
|
||||
model string
|
||||
resume string
|
||||
}
|
||||
|
||||
var scanOpts scanOptions
|
||||
@@ -207,7 +207,7 @@ func executeScan(opts scanOptions) (retErr error) {
|
||||
CommentCollector: rt.Collector,
|
||||
CommentWorkerPool: llmloop.NewCommentWorkerPool(opts.concurrency),
|
||||
MaxConcurrency: opts.concurrency,
|
||||
ConcurrentTaskTimeout: opts.perFileTimeout,
|
||||
ConcurrentTaskTimeout: opts.concurrentTaskTimeout,
|
||||
Model: rt.Model,
|
||||
Background: opts.background,
|
||||
GitRunner: cc.GitRunner,
|
||||
|
||||
@@ -276,8 +276,8 @@ func TestParseScanFlags_IntFlags(t *testing.T) {
|
||||
if opts.concurrency != 16 {
|
||||
t.Errorf("concurrency = %d", opts.concurrency)
|
||||
}
|
||||
if opts.perFileTimeout != 20 {
|
||||
t.Errorf("perFileTimeout = %d", opts.perFileTimeout)
|
||||
if opts.concurrentTaskTimeout != 20 {
|
||||
t.Errorf("concurrentTaskTimeout = %d", opts.concurrentTaskTimeout)
|
||||
}
|
||||
if opts.maxTools != 50 {
|
||||
t.Errorf("maxTools = %d", opts.maxTools)
|
||||
|
||||
@@ -50,11 +50,11 @@ func addExcludeFlag(cmd *cobra.Command, target *string) {
|
||||
}
|
||||
|
||||
func addConcurrencyFlags(cmd *cobra.Command, concurrency, timeout, maxTools, maxGitProcs, maxTokens, maxTokensBudget *int) {
|
||||
cmd.Flags().IntVar(concurrency, "concurrency", 8, "max concurrent file reviews")
|
||||
cmd.Flags().IntVar(concurrency, "concurrency", 8, "max concurrent file-group reviews")
|
||||
cmd.Flags().IntVar(timeout, "timeout", 15, "concurrent task timeout in minutes")
|
||||
cmd.Flags().IntVar(maxTools, "max-tools", 0, "max tool call rounds per subtask (0 = template default; min 50)")
|
||||
cmd.Flags().IntVar(maxGitProcs, "max-git-procs", 16, "max concurrent git subprocesses")
|
||||
cmd.Flags().IntVar(maxTokens, "max-tokens", 0, "per-file prompt token ceiling (0 = configured or template default)")
|
||||
cmd.Flags().IntVar(maxTokens, "max-tokens", 0, "per-group prompt token ceiling (0 = configured or template default)")
|
||||
cmd.Flags().IntVar(maxTokensBudget, "max-tokens-budget", 0, "cap total token usage (input+output) for this review; dispatch stops once exceeded and skipped files are reported as failed(budget). Partial results are published and review exits 0; it exits non-zero only if every selected item failed (0 = unlimited)")
|
||||
}
|
||||
|
||||
@@ -209,7 +209,7 @@ func registerReviewFlags(cmd *cobra.Command, opts *reviewOptions) {
|
||||
addExcludeFlag(cmd, &opts.excludes)
|
||||
addOutputFlags(cmd, &opts.outputFormat, &opts.audience)
|
||||
addOutputPathFlag(cmd, &opts.outputPath)
|
||||
addConcurrencyFlags(cmd, &opts.concurrency, &opts.perFileTimeout, &opts.maxTools, &opts.maxGitProcs, &opts.maxTokens, &opts.maxTokensBudget)
|
||||
addConcurrencyFlags(cmd, &opts.concurrency, &opts.concurrentTaskTimeout, &opts.maxTools, &opts.maxGitProcs, &opts.maxTokens, &opts.maxTokensBudget)
|
||||
addBackgroundFlags(cmd, &opts.background, &opts.backgroundFile)
|
||||
addProviderFlag(cmd, &opts.provider)
|
||||
addModelFlag(cmd, &opts.model)
|
||||
@@ -229,7 +229,7 @@ func registerScanFlags(cmd *cobra.Command, opts *scanOptions) {
|
||||
addOutputFlags(cmd, &opts.outputFormat, &opts.audience)
|
||||
addOutputPathFlag(cmd, &opts.outputPath)
|
||||
cmd.Flags().IntVar(&opts.concurrency, "concurrency", 8, "max concurrent file scans")
|
||||
cmd.Flags().IntVar(&opts.perFileTimeout, "timeout", 15, "concurrent task timeout in minutes")
|
||||
cmd.Flags().IntVar(&opts.concurrentTaskTimeout, "timeout", 15, "concurrent task timeout in minutes")
|
||||
cmd.Flags().IntVar(&opts.maxTools, "max-tools", 0, "max tool call rounds per file; only takes effect when greater than template default")
|
||||
cmd.Flags().IntVar(&opts.maxGitProcs, "max-git-procs", 16, "max concurrent git subprocesses")
|
||||
cmd.Flags().IntVar(&opts.maxTokens, "max-tokens", 0, "per-file prompt token ceiling (0 = configured or template default)")
|
||||
|
||||
+10
-10
@@ -101,7 +101,7 @@ type Args struct {
|
||||
// executeToolCall instead of via a separate worker pool.
|
||||
CommentWorkerPool *CommentWorkerPool
|
||||
|
||||
// Concurrency limit for per-file subtasks. MaxConcurrency <= 0 defaults to 8.
|
||||
// Concurrency limit for per-group subtasks. MaxConcurrency <= 0 defaults to 8.
|
||||
MaxConcurrency int
|
||||
|
||||
// Concurrent task timeout in minutes. 0 means no timeout.
|
||||
@@ -146,7 +146,7 @@ type Args struct {
|
||||
SealedInput *diff.InputResolution
|
||||
|
||||
// MaxTokensBudget caps the aggregate token usage (input+output) across the
|
||||
// whole run; dispatch stops once the running total + a per-file look-ahead
|
||||
// whole run; dispatch stops once the running total + a per-group look-ahead
|
||||
// would exceed it. 0 = unlimited. Mirrors scan.Args.MaxTokensBudget.
|
||||
MaxTokensBudget int64
|
||||
|
||||
@@ -177,7 +177,7 @@ type RuntimeConfig struct {
|
||||
|
||||
// Agent orchestrates the AI-powered code review. LLM tool-use loop / memory
|
||||
// compression / token aggregation now live in internal/llmloop.Runner; this
|
||||
// struct holds the diff-side state and orchestrates per-file subtasks.
|
||||
// struct holds the diff-side state and orchestrates per-group subtasks.
|
||||
type Agent struct {
|
||||
args Args
|
||||
diffs []model.Diff // parsed diffs
|
||||
@@ -274,10 +274,10 @@ func (a *Agent) newRequestMeta(filePath string, taskType session.TaskType, reque
|
||||
}
|
||||
}
|
||||
|
||||
// Run executes the full review pipeline: parse diffs -> plan per file -> LLM tool-loop -> collect comments.
|
||||
// Run executes the full review pipeline: parse diffs -> group -> plan per group -> LLM tool-loop -> collect comments.
|
||||
func (a *Agent) Run(ctx context.Context) ([]model.LlmComment, error) {
|
||||
// Base prompt-cache affinity key for any LLM request in this run that a task doesn't re-scope.
|
||||
// Each task conversation (plan, per-file main loop, compression, ...) refines it with llm.SessionTaskKey where it starts,
|
||||
// Each task conversation (plan, per-group main loop, compression, ...) refines it with llm.SessionTaskKey where it starts,
|
||||
// so affinity keys stay per-conversation, the granularity provider prompt caches actually reuse prefixes at.
|
||||
ctx = llm.ContextWithSessionKey(ctx, a.SessionID())
|
||||
|
||||
@@ -373,7 +373,7 @@ func (a *Agent) Run(ctx context.Context) ([]model.LlmComment, error) {
|
||||
a.session.RecordResumeLineage(session.NewResumeLineage(
|
||||
a.args.Resume, a.session.SessionID, a.args.Provider, a.args.Model))
|
||||
|
||||
// Step 2: Dispatch per-file subtasks concurrently
|
||||
// Step 2: Dispatch per-group subtasks concurrently
|
||||
comments, err := a.dispatchSubtasks(ctx)
|
||||
if len(comments) > 0 {
|
||||
telemetry.RecordCommentsGenerated(ctx, int64(len(comments)))
|
||||
@@ -586,7 +586,7 @@ func (a *Agent) injectDiffMap() {
|
||||
}
|
||||
}
|
||||
|
||||
// dispatchSubtasks runs the Plan + Main phases for each changed file concurrently.
|
||||
// dispatchSubtasks runs the Plan + Main phases for each file group concurrently.
|
||||
func (a *Agent) dispatchSubtasks(ctx context.Context) ([]model.LlmComment, error) {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
@@ -1161,7 +1161,7 @@ var errMainTaskEmpty = errors.New("main_task.messages is empty in template")
|
||||
// safe, generic reason. It never returns the raw error text (which may embed a
|
||||
// provider payload, credentials or absolute paths); the full error is persisted
|
||||
// separately in the session checkpoint. Context deadline/cancel are recognized
|
||||
// via errors.Is (the per-file timeout is the only deadline in play), and the
|
||||
// via errors.Is (the per-group subtask timeout is the only deadline in play), and the
|
||||
// empty-template precondition is a configuration failure.
|
||||
func classifyItemError(err error) (session.FailureClass, string) {
|
||||
switch {
|
||||
@@ -1446,7 +1446,7 @@ func (a *Agent) executeGroupSubtask(ctx context.Context, g FileGroup) (bool, *su
|
||||
defer mainSpan.End()
|
||||
telemetry.SetAttr(mainSpan, "group.label", groupKey)
|
||||
telemetry.SetAttr(mainSpan, "round", round)
|
||||
completed, stop, err := a.runner.RunPerFile(ctx, messages, groupKey)
|
||||
completed, stop, err := a.runner.RunMainTask(ctx, messages, groupKey)
|
||||
if err != nil {
|
||||
mainSpan.SetStatus(codes.Error, err.Error())
|
||||
mainSpan.RecordError(err)
|
||||
@@ -2154,7 +2154,7 @@ func orderedToolParameters(raw json.RawMessage) ([]orderedToolParameter, bool) {
|
||||
}
|
||||
|
||||
// allDiffs exposes the reviewed diff set for cross-file comment re-filing.
|
||||
// It is read-only and safe to call from the per-file subtask goroutines: every
|
||||
// It is read-only and safe to call from the per-group subtask goroutines: every
|
||||
// mutation of a.diffs (filterDiffs, filterLargeDiffs) completes before dispatch
|
||||
// begins, so the slice is stable for the rest of the run.
|
||||
func (a *Agent) allDiffs() []model.Diff {
|
||||
|
||||
@@ -50,8 +50,8 @@ type Estimate struct {
|
||||
// but counts tokens of the diff text (d.Diff) rather than whole-file content,
|
||||
// since the diff path reviews patches, not full files. Returns 0 for deleted
|
||||
// files (they are skipped before dispatch and must not trip the gate). Used
|
||||
// both by the aggregate estimate (estimateDiffCost) and by the per-file budget
|
||||
// look-ahead in dispatchSubtasks.
|
||||
// both by the aggregate estimate (estimateDiffCost) and by the per-group budget
|
||||
// look-ahead in dispatchSubtasks, which sums this over a group's diffs.
|
||||
func estimateDiffFileTokens(d model.Diff) int64 {
|
||||
if d.IsDeleted || d.Diff == "" {
|
||||
return 0
|
||||
|
||||
@@ -528,7 +528,7 @@ func TestManifestFlowGroupBudgetStopIsPartialWhenSomeFilesHaveComments(t *testin
|
||||
// llmloop.Runner snapshots Template by value at construction time — mutating
|
||||
// a.args.Template afterward (as other tests in this file do for MaxTokens,
|
||||
// which executeGroupSubtask re-reads from a.args.Template directly) would
|
||||
// not reach the runner's copy that RunPerFile actually consumes.
|
||||
// not reach the runner's copy that RunMainTask actually consumes.
|
||||
a := New(Args{
|
||||
RepoDir: repoDir,
|
||||
From: "main",
|
||||
|
||||
@@ -54,10 +54,10 @@ type compressionJob struct {
|
||||
}
|
||||
|
||||
// compressionState is the async-compression bookkeeping for a single
|
||||
// conversation (one RunPerFile call). The Runner is shared by concurrent
|
||||
// per-file goroutines, so this state must not live on the Runner: a shared
|
||||
// slot lets one file apply, cancel, or replace another file's compression
|
||||
// job (#384).
|
||||
// conversation (one RunMainTask call). The Runner is shared by concurrent
|
||||
// per-subtask goroutines, so this state must not live on the Runner: a shared
|
||||
// slot lets one subtask apply, cancel, or replace another subtask's
|
||||
// compression job (#384).
|
||||
type compressionState struct {
|
||||
mu sync.Mutex
|
||||
pendingJob *compressionJob
|
||||
@@ -220,7 +220,7 @@ func copyMessages(msgs []llm.Message) []llm.Message {
|
||||
// messages, summarizing the compress zone while preserving the active zone
|
||||
// intact. Returns rebuilt as [frozen] + [compressed_summary appended to
|
||||
// the user prompt] + [active].
|
||||
func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePath string) ([]llm.Message, error) {
|
||||
func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, taskKey string) ([]llm.Message, error) {
|
||||
if len(r.deps.Template.MemoryCompressionTask.Messages) == 0 || len(msgs) <= 2 {
|
||||
return msgs[:min(len(msgs), 2)], nil
|
||||
}
|
||||
@@ -244,14 +244,14 @@ func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePat
|
||||
// llm_request line reaches the session JSONL before the response: a run
|
||||
// killed mid-request now leaves an llm_request with no response, which
|
||||
// resume ignores (applyResumeLine has no case for it).
|
||||
fs := r.deps.Session.GetOrCreateFileSession(filePath)
|
||||
fs := r.deps.Session.GetOrCreateFileSession(taskKey)
|
||||
rec := fs.AppendTaskRecord(session.MemoryCompressionTask, compressionMsgs)
|
||||
|
||||
ctx = llm.ContextWithSessionKey(ctx,
|
||||
llm.SessionTaskKey(r.deps.Session.SessionID, string(session.MemoryCompressionTask), filePath))
|
||||
llm.SessionTaskKey(r.deps.Session.SessionID, string(session.MemoryCompressionTask), taskKey))
|
||||
|
||||
startTime := time.Now()
|
||||
reqCtx := r.requestCtx(ctx, filePath, session.MemoryCompressionTask, rec.RequestNo)
|
||||
reqCtx := r.requestCtx(ctx, taskKey, session.MemoryCompressionTask, rec.RequestNo)
|
||||
resp, err := r.deps.LLMClient.CompletionsWithCtx(reqCtx, llm.ChatRequest{
|
||||
Model: r.deps.Model,
|
||||
Messages: compressionMsgs,
|
||||
@@ -299,7 +299,7 @@ func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePat
|
||||
// conversation owning st. A no-op when a job is already pending — the
|
||||
// check-and-set happens under st.mu so concurrent callers cannot replace
|
||||
// (and thereby leak) an in-flight job.
|
||||
func (r *Runner) triggerAsyncCompression(ctx context.Context, st *compressionState, messages []llm.Message, filePath string) {
|
||||
func (r *Runner) triggerAsyncCompression(ctx context.Context, st *compressionState, messages []llm.Message, taskKey string) {
|
||||
st.mu.Lock()
|
||||
if st.pendingJob != nil {
|
||||
st.mu.Unlock()
|
||||
@@ -317,7 +317,7 @@ func (r *Runner) triggerAsyncCompression(ctx context.Context, st *compressionSta
|
||||
go func() {
|
||||
defer r.bg.Done()
|
||||
defer cancel()
|
||||
rebuilt, err := r.runCompression(asyncCtx, msgSnapshot, filePath)
|
||||
rebuilt, err := r.runCompression(asyncCtx, msgSnapshot, taskKey)
|
||||
|
||||
st.mu.Lock()
|
||||
defer st.mu.Unlock()
|
||||
|
||||
+84
-57
@@ -52,7 +52,7 @@ type Deps struct {
|
||||
|
||||
// NewRequestMeta builds the retry-report identity for one logical LLM
|
||||
// request. Non-nil only for review: the retry report describes ocr review,
|
||||
// and this Runner is shared with scan (internal/scan.Agent calls RunPerFile),
|
||||
// and this Runner is shared with scan (internal/scan.Agent calls RunMainTask),
|
||||
// so main_task, memory compression and re-location all run under both modes.
|
||||
//
|
||||
// The gate has to be this field rather than a Provider string, because an
|
||||
@@ -68,19 +68,27 @@ type Deps struct {
|
||||
// requestCtx returns ctx carrying the identity of one logical LLM request, or
|
||||
// ctx unchanged when identity is disabled (scan) or the meta is unusable.
|
||||
//
|
||||
// taskKey feeds NewRequestMeta's filePath parameter, and only review ever
|
||||
// reaches that call — scan leaves the factory nil. What arrives there is still
|
||||
// not always a file path: the main-task and memory-compression callers pass the
|
||||
// subtask key, which for a group of several files is the group key, while the
|
||||
// re-location caller passes the comment's own path. The parameter keeps the name
|
||||
// filePath because the session records it feeds spell the slot that way, so the
|
||||
// two names differ on purpose rather than by oversight.
|
||||
//
|
||||
// Callers must invoke it after AppendTaskRecord and pass that record's
|
||||
// RequestNo — the fixed order is AppendTaskRecord -> requestCtx ->
|
||||
// CompletionsWithCtx -> SetResponse/SetError.
|
||||
func (r *Runner) requestCtx(ctx context.Context, filePath string, taskType session.TaskType, requestNo int) context.Context {
|
||||
func (r *Runner) requestCtx(ctx context.Context, taskKey string, taskType session.TaskType, requestNo int) context.Context {
|
||||
if r.deps.NewRequestMeta == nil {
|
||||
return ctx
|
||||
}
|
||||
return llm.WithRequestMeta(ctx, r.deps.NewRequestMeta(filePath, taskType, requestNo))
|
||||
return llm.WithRequestMeta(ctx, r.deps.NewRequestMeta(taskKey, taskType, requestNo))
|
||||
}
|
||||
|
||||
// Runner is a per-session (across files) executor of the LLM tool-use
|
||||
// loop. Token counters and warnings are aggregated across every RunPerFile
|
||||
// call; background memory compression is scoped to each RunPerFile
|
||||
// Runner is a per-session (across subtasks) executor of the LLM tool-use
|
||||
// loop. Token counters and warnings are aggregated across every RunMainTask
|
||||
// call; background memory compression is scoped to each RunMainTask
|
||||
// conversation (see compressionState).
|
||||
type Runner struct {
|
||||
deps Deps
|
||||
@@ -95,7 +103,7 @@ type Runner struct {
|
||||
toolCallSequence int64
|
||||
toolFailures []ToolFailureDetail
|
||||
// bg tracks every background goroutine that can still issue an LLM
|
||||
// request after RunPerFile returned. WaitBackground joins them so a
|
||||
// request after RunMainTask returned. WaitBackground joins them so a
|
||||
// retry-report Freeze at the run boundary cannot observe an
|
||||
// un-finalized request. See WaitBackground.
|
||||
bg sync.WaitGroup
|
||||
@@ -105,7 +113,11 @@ type Runner struct {
|
||||
type ToolFailureDetail struct {
|
||||
ToolCallNumber int64 `json:"tool_call_number"`
|
||||
ToolName string `json:"tool_name"`
|
||||
FilePath string `json:"file_path,omitempty"`
|
||||
// FilePath carries the failing subtask's key, which is a file path only in
|
||||
// scan; a review group of several files reports its group key here. The
|
||||
// field reaches --format json as file_path, so a consumer must not assume
|
||||
// it parses as a single path.
|
||||
FilePath string `json:"file_path,omitempty"`
|
||||
// Arguments is the raw tool-call argument string returned by the LLM.
|
||||
Arguments string `json:"arguments"`
|
||||
Error string `json:"error"`
|
||||
@@ -119,13 +131,13 @@ func NewRunner(deps Deps) *Runner {
|
||||
// WaitBackground blocks until every background job started by this Runner has
|
||||
// returned. Background memory compression is the only such job, and
|
||||
// cancelPendingCompression cancels it without waiting — its goroutine can
|
||||
// therefore still be inside an LLM request after RunPerFile returned. Callers
|
||||
// therefore still be inside an LLM request after RunMainTask returned. Callers
|
||||
// that freeze a retry report at the run boundary must join here first:
|
||||
// RetryCollector.Freeze rejects any request that has not been finalized and
|
||||
// discards the whole report, which would otherwise be an intermittent race.
|
||||
//
|
||||
// Every pending job has already been cancelled by the time the last
|
||||
// RunPerFile returns (cancelPendingCompression runs as a deferred call on
|
||||
// RunMainTask returns (cancelPendingCompression runs as a deferred call on
|
||||
// every exit, and triggerAsyncCompression refuses to start a second job while
|
||||
// one is pending), so this normally returns quickly — but the wait length
|
||||
// ultimately depends on the LLM client honouring context cancellation, and no
|
||||
@@ -208,12 +220,12 @@ func (r *Runner) recordToolCall(name string) int64 {
|
||||
return r.toolCallSequence
|
||||
}
|
||||
|
||||
func (r *Runner) recordToolFailure(number int64, name, filePath, errMsg string,
|
||||
func (r *Runner) recordToolFailure(number int64, name, taskKey, errMsg string,
|
||||
rec *session.TaskRecord, rawArguments string, duration time.Duration) {
|
||||
detail := ToolFailureDetail{
|
||||
ToolCallNumber: number,
|
||||
ToolName: name,
|
||||
FilePath: filePath,
|
||||
FilePath: taskKey,
|
||||
Arguments: rawArguments,
|
||||
Error: errMsg,
|
||||
}
|
||||
@@ -230,7 +242,7 @@ func (r *Runner) recordToolFailure(number int64, name, filePath, errMsg string,
|
||||
// RecordUsage adds the prompt/completion/cache tokens reported by an LLM
|
||||
// response to the runner's aggregate counters. Used by callers (plan phase
|
||||
// in agent / future scan phases) that perform their own LLM calls outside
|
||||
// RunPerFile.
|
||||
// RunMainTask.
|
||||
func (r *Runner) RecordUsage(u *llm.UsageInfo) {
|
||||
if u == nil {
|
||||
return
|
||||
@@ -251,7 +263,7 @@ func (r *Runner) CollectPendingComments() []model.LlmComment {
|
||||
return r.deps.CommentCollector.Comments()
|
||||
}
|
||||
|
||||
// MainLoopStop classifies why RunPerFile stopped without an explicit task_done
|
||||
// MainLoopStop classifies why RunMainTask stopped without an explicit task_done
|
||||
// and without a Go error. It lets the caller attribute a precise, honest failure
|
||||
// classification instead of guessing from free text: only a configured limit
|
||||
// (max tool-request rounds) is a budget stop; the empty-round and compression
|
||||
@@ -260,7 +272,7 @@ func (r *Runner) CollectPendingComments() []model.LlmComment {
|
||||
type MainLoopStop int
|
||||
|
||||
const (
|
||||
// StopNone — RunPerFile completed via task_done or returned an error; the
|
||||
// StopNone — RunMainTask completed via task_done or returned an error; the
|
||||
// stop cause carries no additional meaning.
|
||||
StopNone MainLoopStop = iota
|
||||
// StopMaxRounds — the configured MaxToolRequestTimes round budget was
|
||||
@@ -321,21 +333,31 @@ func (s MainLoopStop) Reason() string {
|
||||
}
|
||||
}
|
||||
|
||||
// RunPerFile drives the main LLM conversation loop for a single file.
|
||||
// It sends messages with the configured tool definitions, executes any
|
||||
// tool calls returned by the model, and collects review comments until
|
||||
// task_done is called or limits are reached. Token usage and warnings
|
||||
// are aggregated on the Runner across all files. The returned bool is true
|
||||
// only when the model explicitly calls task_done with a successful state. The
|
||||
// MainLoopStop return classifies a non-completed, non-error stop at its trigger
|
||||
// point so the caller never has to infer the cause from text or context state.
|
||||
func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath string) (bool, MainLoopStop, error) {
|
||||
// RunMainTask drives one MAIN_TASK conversation loop to its end. It sends
|
||||
// messages with the configured tool definitions, executes any tool calls
|
||||
// returned by the model, and collects review comments until task_done is
|
||||
// called or limits are reached. Token usage and warnings are aggregated on the
|
||||
// Runner across every call. The returned bool is true only when the model
|
||||
// explicitly calls task_done with a successful state. The MainLoopStop return
|
||||
// classifies a non-completed, non-error stop at its trigger point so the caller
|
||||
// never has to infer the cause from text or context state.
|
||||
//
|
||||
// taskKey identifies the subtask this conversation belongs to, and is not
|
||||
// necessarily a file path: scan passes one file's path, while review passes the
|
||||
// group key of a file group that may hold several files. It is used for cache
|
||||
// affinity, session bookkeeping, progress lines and as the fallback path for a
|
||||
// code_comment the model filed without one — that last use is why a group key
|
||||
// reaching here matters, and why callers should keep it human-readable.
|
||||
//
|
||||
// Review calls this once per review round, so one review subtask can drive
|
||||
// several of these conversations in sequence.
|
||||
func (r *Runner) RunMainTask(ctx context.Context, messages []llm.Message, taskKey string) (bool, MainLoopStop, error) {
|
||||
// Every round of this loop re-sends the growing conversation, so each
|
||||
// request is a prefix extension of the previous one — exactly what
|
||||
// provider prompt caches reuse. Scope the affinity key to this file's
|
||||
// provider prompt caches reuse. Scope the affinity key to this subtask's
|
||||
// main-task conversation so every round routes to the same cache node.
|
||||
ctx = llm.ContextWithSessionKey(ctx,
|
||||
llm.SessionTaskKey(r.deps.Session.SessionID, string(session.MainTask), newPath))
|
||||
llm.SessionTaskKey(r.deps.Session.SessionID, string(session.MainTask), taskKey))
|
||||
|
||||
toolReqCount := r.deps.Template.MaxToolRequestTimes
|
||||
const maxConsecutiveEmptyRounds = 3
|
||||
@@ -360,13 +382,13 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
|
||||
|
||||
toolReqCount--
|
||||
|
||||
fs := r.deps.Session.GetOrCreateFileSession(newPath)
|
||||
fs := r.deps.Session.GetOrCreateFileSession(taskKey)
|
||||
rec := fs.AppendTaskRecord(session.MainTask, append([]llm.Message(nil), messages...))
|
||||
startTime := time.Now()
|
||||
|
||||
// Scoped to this round: ctx itself must stay identity-free so each
|
||||
// iteration's meta replaces the previous one instead of nesting.
|
||||
reqCtx := r.requestCtx(ctx, newPath, session.MainTask, rec.RequestNo)
|
||||
reqCtx := r.requestCtx(ctx, taskKey, session.MainTask, rec.RequestNo)
|
||||
|
||||
_, llmSpan := telemetry.StartLLMSpan(ctx, r.deps.Model)
|
||||
resp, err := r.deps.LLMClient.CompletionsWithCtx(reqCtx, llm.ChatRequest{
|
||||
@@ -401,7 +423,7 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
|
||||
calls := resp.ToolCalls()
|
||||
|
||||
if len(calls) == 0 {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] No tool calls parsed for %s, retrying...\n", newPath)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] No tool calls parsed for %s, retrying...\n", taskKey)
|
||||
messages = append(messages, llm.NewTextMessage("user", "You did not successfully call any tools. Please try again or use task_done if finished."))
|
||||
native := resp.Native()
|
||||
reasoning := resp.ReasoningContent()
|
||||
@@ -420,7 +442,7 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
|
||||
// Reasoning is turn-level, so all tool calls in this turn share it.
|
||||
thinking := resp.ReasoningContent()
|
||||
for _, call := range calls {
|
||||
cp := r.executeToolCall(ctx, newPath, call, rec, thinking)
|
||||
cp := r.executeToolCall(ctx, taskKey, call, rec, thinking)
|
||||
if cp.Failed {
|
||||
return false, StopNone, fmt.Errorf("task failed: %s", cp.Data)
|
||||
} else if cp.Completed {
|
||||
@@ -452,26 +474,26 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
|
||||
if !hasValidResult {
|
||||
consecutiveEmptyRounds++
|
||||
if consecutiveEmptyRounds >= maxConsecutiveEmptyRounds {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Too many empty retries for %s, stopping.\n", newPath)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Too many empty retries for %s, stopping.\n", taskKey)
|
||||
stop = StopEmptyRounds
|
||||
break
|
||||
}
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] No valid tool results for %s, retrying...\n", newPath)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] No valid tool results for %s, retrying...\n", taskKey)
|
||||
} else {
|
||||
consecutiveEmptyRounds = 0
|
||||
}
|
||||
|
||||
succeed := r.addNextMessage(ctx, content, calls, resp.Native(), thinking, results, &messages, newPath, st)
|
||||
succeed := r.addNextMessage(ctx, content, calls, resp.Native(), thinking, results, &messages, taskKey, st)
|
||||
if !succeed {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Context compression exceeded threshold for %s, stopping.\n", newPath)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Context compression exceeded threshold for %s, stopping.\n", taskKey)
|
||||
stop = StopCompression
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if stop == StopMaxRounds {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Max tool requests reached for %s.\n", newPath)
|
||||
r.runGraceRound(ctx, messages, newPath, sessionID)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Max tool requests reached for %s.\n", taskKey)
|
||||
r.runGraceRound(ctx, messages, taskKey, sessionID)
|
||||
}
|
||||
return false, stop, nil
|
||||
}
|
||||
@@ -479,7 +501,7 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
|
||||
// runGraceRound performs one final LLM call after the tool-request budget is
|
||||
// exhausted, giving the model a chance to submit any findings it identified
|
||||
// but did not yet report via code_comment.
|
||||
func (r *Runner) runGraceRound(ctx context.Context, messages []llm.Message, newPath string, sessionID string) {
|
||||
func (r *Runner) runGraceRound(ctx context.Context, messages []llm.Message, taskKey string, sessionID string) {
|
||||
graceDefs := graceRoundToolDefs(r.deps.MainToolDefs)
|
||||
if len(graceDefs) == 0 {
|
||||
return
|
||||
@@ -492,14 +514,14 @@ func (r *Runner) runGraceRound(ctx context.Context, messages []llm.Message, newP
|
||||
"No other tools are available. Do not attempt further analysis."))
|
||||
|
||||
if ctx.Err() != nil {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Grace round skipped for %s: context cancelled\n", newPath)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Grace round skipped for %s: context cancelled\n", taskKey)
|
||||
return
|
||||
}
|
||||
|
||||
fs := r.deps.Session.GetOrCreateFileSession(newPath)
|
||||
fs := r.deps.Session.GetOrCreateFileSession(taskKey)
|
||||
rec := fs.AppendTaskRecord(session.MainTask, messages)
|
||||
startTime := time.Now()
|
||||
reqCtx := r.requestCtx(ctx, newPath, session.MainTask, rec.RequestNo)
|
||||
reqCtx := r.requestCtx(ctx, taskKey, session.MainTask, rec.RequestNo)
|
||||
|
||||
_, llmSpan := telemetry.StartLLMSpan(ctx, r.deps.Model)
|
||||
resp, err := r.deps.LLMClient.CompletionsWithCtx(reqCtx, llm.ChatRequest{
|
||||
@@ -515,7 +537,7 @@ func (r *Runner) runGraceRound(ctx context.Context, messages []llm.Message, newP
|
||||
telemetry.RecordLLMResult(llmSpan, duration, 0, err)
|
||||
llmSpan.End()
|
||||
telemetry.RecordLLMRequest(ctx, r.deps.Model, duration, 0, "error")
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Grace round LLM error for %s: %v\n", newPath, err)
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Grace round LLM error for %s: %v\n", taskKey, err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -539,7 +561,7 @@ func (r *Runner) runGraceRound(ctx context.Context, messages []llm.Message, newP
|
||||
|
||||
thinking := resp.ReasoningContent()
|
||||
for _, call := range calls {
|
||||
r.executeToolCall(ctx, newPath, call, rec, thinking)
|
||||
r.executeToolCall(ctx, taskKey, call, rec, thinking)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -559,7 +581,11 @@ func graceRoundToolDefs(defs []llm.ToolDef) []llm.ToolDef {
|
||||
// records the result in session history. code_comment handling includes
|
||||
// optional async dispatch through CommentWorkerPool plus line-number
|
||||
// resolution / re-location.
|
||||
func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.ToolCall, rec *session.TaskRecord, thinking string) tool.TaskCheckpoint {
|
||||
//
|
||||
// taskKey is the calling subtask's key (see RunMainTask). It labels the
|
||||
// session/telemetry records and the worker-pool submission, and stands in as a
|
||||
// code_comment's path when the model omitted one.
|
||||
func (r *Runner) executeToolCall(ctx context.Context, taskKey string, call llm.ToolCall, rec *session.TaskRecord, thinking string) tool.TaskCheckpoint {
|
||||
t := tool.OfName(call.Function.Name)
|
||||
|
||||
if t == tool.TaskDone {
|
||||
@@ -599,7 +625,7 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
errMsg := fmt.Sprintf("Error parsing tool arguments for %s: %v", toolName, err)
|
||||
telemetry.PrintToolCallStarted(toolName, nil)
|
||||
telemetry.PrintToolCallError(toolName, fmt.Errorf("%s", errMsg))
|
||||
r.recordToolFailure(toolCallNumber, toolName, newPath, errMsg,
|
||||
r.recordToolFailure(toolCallNumber, toolName, taskKey, errMsg,
|
||||
rec, call.Function.Arguments, time.Since(callStarted))
|
||||
return tool.Of(errMsg)
|
||||
}
|
||||
@@ -610,12 +636,12 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
telemetry.PrintToolCallStarted(t.Name(), args)
|
||||
_, toolSpan := telemetry.StartToolSpan(ctx, t.Name())
|
||||
|
||||
comments, repair, errMsg := tool.ParseCommentsWithPath(args, newPath)
|
||||
comments, repair, errMsg := tool.ParseCommentsWithPath(args, taskKey)
|
||||
if repair != nil {
|
||||
// The model sees a plain success, so this warning is the only record
|
||||
// that its `comments` violated the array schema — without it the
|
||||
// repair would absorb an unbounded number of them unobserved.
|
||||
r.RecordWarning("comment_args_repaired", newPath, repair.Message())
|
||||
r.RecordWarning("comment_args_repaired", taskKey, repair.Message())
|
||||
}
|
||||
if errMsg != "" {
|
||||
dur := time.Since(startTime)
|
||||
@@ -623,7 +649,7 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
telemetry.RecordToolResult(toolSpan, t.Name(), dur.Milliseconds(), toolErr)
|
||||
toolSpan.End()
|
||||
telemetry.RecordToolCall(ctx, t.Name(), dur, false)
|
||||
r.recordToolFailure(toolCallNumber, toolName, newPath, errMsg,
|
||||
r.recordToolFailure(toolCallNumber, toolName, taskKey, errMsg,
|
||||
rec, call.Function.Arguments, dur)
|
||||
telemetry.PrintToolCallError(t.Name(), toolErr)
|
||||
return tool.Of(errMsg)
|
||||
@@ -673,10 +699,11 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
rlRec := fs.AppendTaskRecord(session.ReLocationTask, msgs)
|
||||
// FilePath is cm.Path so it cannot drift from the file
|
||||
// session opened above — that join is what the report
|
||||
// needs. It equals newPath whenever newPath is set,
|
||||
// because the path arg is overridden with it further
|
||||
// up, but reading it from the comment keeps the two
|
||||
// aligned without depending on that.
|
||||
// needs. Reading it from the comment rather than from
|
||||
// taskKey is what makes this correct for a grouped
|
||||
// review: taskKey is only the comment's fallback path,
|
||||
// so a group of several files resolves each of its
|
||||
// comments under the file the comment names.
|
||||
rlCtx := llm.ContextWithSessionKey(rctx,
|
||||
llm.SessionTaskKey(r.deps.Session.SessionID, string(session.ReLocationTask), cm.Path))
|
||||
reqCtx := r.requestCtx(rlCtx, cm.Path, session.ReLocationTask, rlRec.RequestNo)
|
||||
@@ -706,7 +733,7 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
pool := r.deps.CommentWorkerPool
|
||||
asyncCtx := context.WithoutCancel(ctx)
|
||||
toolName := t.Name()
|
||||
pool.SubmitFor(newPath, func() ([]model.LlmComment, error) {
|
||||
pool.SubmitFor(taskKey, func() ([]model.LlmComment, error) {
|
||||
defer func() {
|
||||
dur := time.Since(startTime)
|
||||
telemetry.RecordToolResult(toolSpan, toolName, dur.Milliseconds(), nil)
|
||||
@@ -743,7 +770,7 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
telemetry.RecordToolCall(ctx, toolName, dur, ok)
|
||||
|
||||
if err != nil {
|
||||
r.recordToolFailure(toolCallNumber, toolName, newPath, err.Error(),
|
||||
r.recordToolFailure(toolCallNumber, toolName, taskKey, err.Error(),
|
||||
rec, call.Function.Arguments, dur)
|
||||
telemetry.PrintToolCallError(toolName, err)
|
||||
return tool.Of(fmt.Sprintf("Error executing tool %s: %v", toolName, err))
|
||||
@@ -760,7 +787,7 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
|
||||
// warning (80%) MaxTokens thresholds. Returns false when even after
|
||||
// synchronous compression the conversation is still over the warning
|
||||
// threshold — caller should stop the loop in that case.
|
||||
func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, toolCalls []llm.ToolCall, native llm.NativeTurn, reasoningContent string, results []tool.ToolCallResult, messages *[]llm.Message, filePath string, st *compressionState) bool {
|
||||
func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, toolCalls []llm.ToolCall, native llm.NativeTurn, reasoningContent string, results []tool.ToolCallResult, messages *[]llm.Message, taskKey string, st *compressionState) bool {
|
||||
maxAllowed := r.deps.Template.MaxTokens
|
||||
softLimit := int(float64(maxAllowed) * tokenSoftThreshold)
|
||||
warnLimit := PromptTokenLimit(maxAllowed)
|
||||
@@ -772,7 +799,7 @@ func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, to
|
||||
if CountMessagesTokens(*messages) > warnLimit {
|
||||
r.cancelPendingCompression(st)
|
||||
var err error
|
||||
if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
|
||||
if *messages, err = r.runCompression(ctx, *messages, taskKey); err != nil {
|
||||
// Compression failed; continue with over-limit messages — the
|
||||
// post-append check below will retry.
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
|
||||
@@ -793,7 +820,7 @@ func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, to
|
||||
if finalCount > warnLimit {
|
||||
r.cancelPendingCompression(st)
|
||||
var err error
|
||||
if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
|
||||
if *messages, err = r.runCompression(ctx, *messages, taskKey); err != nil {
|
||||
fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
|
||||
}
|
||||
finalCount = CountMessagesTokens(*messages)
|
||||
@@ -803,7 +830,7 @@ func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, to
|
||||
// a job is never started and then immediately cancelled by the same
|
||||
// call (#384), and never started when we are about to return false.
|
||||
if finalCount > softLimit && finalCount < warnLimit {
|
||||
r.triggerAsyncCompression(ctx, st, *messages, filePath)
|
||||
r.triggerAsyncCompression(ctx, st, *messages, taskKey)
|
||||
}
|
||||
|
||||
return finalCount < warnLimit
|
||||
|
||||
@@ -35,7 +35,7 @@ func (s *scriptedLLMClient) CompletionsWithCtx(_ context.Context, req llm.ChatRe
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// newThinkingTestRunner builds a Runner wired for a full RunPerFile pass:
|
||||
// newThinkingTestRunner builds a Runner wired for a full RunMainTask pass:
|
||||
// a scripted LLM client, the code_comment tool, and a comment collector.
|
||||
func newThinkingTestRunner(t *testing.T, client llm.LLMClient) (*Runner, *tool.CommentCollector) {
|
||||
t.Helper()
|
||||
@@ -72,20 +72,20 @@ func codeCommentResponse(reasoning, content string) *llm.ChatResponse {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_BackfillsThinkingFromReasoningContent verifies the full
|
||||
// TestRunMainTask_BackfillsThinkingFromReasoningContent verifies the full
|
||||
// wiring: the model's native reasoning_content on a tool-calling turn is
|
||||
// backfilled into the comment's thinking, while the turn's assistant
|
||||
// content is ignored.
|
||||
func TestRunPerFile_BackfillsThinkingFromReasoningContent(t *testing.T) {
|
||||
func TestRunMainTask_BackfillsThinkingFromReasoningContent(t *testing.T) {
|
||||
client := &scriptedLLMClient{responses: []*llm.ChatResponse{
|
||||
codeCommentResponse("native reasoning", "I'll now leave a comment on this file"),
|
||||
taskDoneResponse(),
|
||||
}}
|
||||
r, collector := newThinkingTestRunner(t, client)
|
||||
|
||||
ok, _, err := r.RunPerFile(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
ok, _, err := r.RunMainTask(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("RunPerFile = (%v, err %v), want completed", ok, err)
|
||||
t.Fatalf("RunMainTask = (%v, err %v), want completed", ok, err)
|
||||
}
|
||||
|
||||
comments := collector.Comments()
|
||||
@@ -97,19 +97,19 @@ func TestRunPerFile_BackfillsThinkingFromReasoningContent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_NoFallbackToContent is a regression test: when a turn has
|
||||
// TestRunMainTask_NoFallbackToContent is a regression test: when a turn has
|
||||
// assistant content but no reasoning_content, the comment thinking must stay
|
||||
// empty. It fails if the removed `thinking = content` fallback returns.
|
||||
func TestRunPerFile_NoFallbackToContent(t *testing.T) {
|
||||
func TestRunMainTask_NoFallbackToContent(t *testing.T) {
|
||||
client := &scriptedLLMClient{responses: []*llm.ChatResponse{
|
||||
codeCommentResponse("", "I'll now leave a comment on this file"),
|
||||
taskDoneResponse(),
|
||||
}}
|
||||
r, collector := newThinkingTestRunner(t, client)
|
||||
|
||||
ok, _, err := r.RunPerFile(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
ok, _, err := r.RunMainTask(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("RunPerFile = (%v, err %v), want completed", ok, err)
|
||||
t.Fatalf("RunMainTask = (%v, err %v), want completed", ok, err)
|
||||
}
|
||||
|
||||
comments := collector.Comments()
|
||||
@@ -121,14 +121,14 @@ func TestRunPerFile_NoFallbackToContent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_DoesNotDuplicateReasoningIntoVisibleContent is a regression
|
||||
// TestRunMainTask_DoesNotDuplicateReasoningIntoVisibleContent is a regression
|
||||
// test found in review of the #805 fix: ChatResponse.Content() falls back to
|
||||
// ReasoningContent when there's no visible text — the common shape for an
|
||||
// openai-chat-completions tool-calling turn (content: null, reasoning_content
|
||||
// set). If the loop built history from Content() instead of VisibleContent(),
|
||||
// that same reasoning text would land in both the ordinary content field and
|
||||
// the reasoning_content native payload on replay, doubling it on the wire.
|
||||
func TestRunPerFile_DoesNotDuplicateReasoningIntoVisibleContent(t *testing.T) {
|
||||
func TestRunMainTask_DoesNotDuplicateReasoningIntoVisibleContent(t *testing.T) {
|
||||
emptyContent := ""
|
||||
reasoning := "private reasoning that must not be duplicated"
|
||||
reasoningResp := &llm.ChatResponse{
|
||||
@@ -149,9 +149,9 @@ func TestRunPerFile_DoesNotDuplicateReasoningIntoVisibleContent(t *testing.T) {
|
||||
client := &scriptedLLMClient{responses: []*llm.ChatResponse{reasoningResp, taskDoneResponse()}}
|
||||
r, _ := newThinkingTestRunner(t, client)
|
||||
|
||||
ok, _, err := r.RunPerFile(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
ok, _, err := r.RunMainTask(context.Background(), []llm.Message{msg("user", "review")}, "file.go")
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("RunPerFile = (%v, err %v), want completed", ok, err)
|
||||
t.Fatalf("RunMainTask = (%v, err %v), want completed", ok, err)
|
||||
}
|
||||
|
||||
// requests[1] is the second call: its Messages carry the history built
|
||||
|
||||
@@ -106,18 +106,18 @@ func newTestDeps(client llm.LLMClient) Deps {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_TaskDoneImmediately(t *testing.T) {
|
||||
func TestRunMainTask_TaskDoneImmediately(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{taskDoneResponse()}}
|
||||
deps := newTestDeps(client)
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review this file")}
|
||||
completed, _, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, _, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if !completed {
|
||||
t.Fatal("expected task_done to complete RunPerFile")
|
||||
t.Fatal("expected task_done to complete RunMainTask")
|
||||
}
|
||||
if client.calls != 1 {
|
||||
t.Errorf("expected 1 LLM call, got %d", client.calls)
|
||||
@@ -130,52 +130,52 @@ func TestRunPerFile_TaskDoneImmediately(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_UsesCompletionTokenLimit(t *testing.T) {
|
||||
func TestRunMainTask_UsesCompletionTokenLimit(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{taskDoneResponse()}}
|
||||
deps := newTestDeps(client)
|
||||
deps.Template.MaxTokens = 200000
|
||||
deps.Template.MaxCompletionTokens = 58888
|
||||
runner := NewRunner(deps)
|
||||
|
||||
_, _, err := runner.RunPerFile(
|
||||
_, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review")},
|
||||
"main.go",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if got := client.requests[0].MaxTokens; got != 58888 {
|
||||
t.Fatalf("request MaxTokens = %d, want 58888", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_TaskDoneExplicitDone(t *testing.T) {
|
||||
func TestRunMainTask_TaskDoneExplicitDone(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
taskDoneResponseWithArguments(`{"state":"DONE"}`),
|
||||
}}
|
||||
runner := NewRunner(newTestDeps(client))
|
||||
|
||||
completed, _, err := runner.RunPerFile(
|
||||
completed, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if !completed {
|
||||
t.Fatal("expected task_done DONE to complete RunPerFile")
|
||||
t.Fatal("expected task_done DONE to complete RunMainTask")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_TaskDoneFailed(t *testing.T) {
|
||||
func TestRunMainTask_TaskDoneFailed(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
taskDoneResponseWithArguments(`{"state":"FAILED"}`),
|
||||
}}
|
||||
runner := NewRunner(newTestDeps(client))
|
||||
|
||||
completed, _, err := runner.RunPerFile(
|
||||
completed, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
@@ -184,14 +184,14 @@ func TestRunPerFile_TaskDoneFailed(t *testing.T) {
|
||||
t.Fatalf("expected task_done FAILED error, got %v", err)
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("task_done FAILED must not complete RunPerFile")
|
||||
t.Fatal("task_done FAILED must not complete RunMainTask")
|
||||
}
|
||||
if client.calls != 1 {
|
||||
t.Fatalf("expected terminal failure after 1 LLM call, got %d", client.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_InvalidTaskDoneStateRetries(t *testing.T) {
|
||||
func TestRunMainTask_InvalidTaskDoneStateRetries(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
arguments string
|
||||
@@ -210,13 +210,13 @@ func TestRunPerFile_InvalidTaskDoneStateRetries(t *testing.T) {
|
||||
}}
|
||||
runner := NewRunner(newTestDeps(client))
|
||||
|
||||
completed, _, err := runner.RunPerFile(
|
||||
completed, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if !completed {
|
||||
t.Fatal("expected retry to complete with task_done DONE")
|
||||
@@ -228,7 +228,7 @@ func TestRunPerFile_InvalidTaskDoneStateRetries(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_TagsRequestsWithTaskSessionKey(t *testing.T) {
|
||||
func TestRunMainTask_TagsRequestsWithTaskSessionKey(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
taskDoneResponse(),
|
||||
@@ -237,8 +237,8 @@ func TestRunPerFile_TagsRequestsWithTaskSessionKey(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review this file")}
|
||||
if _, _, err := runner.RunPerFile(context.Background(), msgs, "main.go"); err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
if _, _, err := runner.RunMainTask(context.Background(), msgs, "main.go"); err != nil {
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
|
||||
want := llm.SessionTaskKey(deps.Session.SessionID, string(session.MainTask), "main.go")
|
||||
@@ -252,7 +252,7 @@ func TestRunPerFile_TagsRequestsWithTaskSessionKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_ToolCallThenDone(t *testing.T) {
|
||||
func TestRunMainTask_ToolCallThenDone(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
taskDoneResponse(),
|
||||
@@ -261,12 +261,12 @@ func TestRunPerFile_ToolCallThenDone(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, _, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, _, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if !completed {
|
||||
t.Fatal("expected task_done to complete RunPerFile")
|
||||
t.Fatal("expected task_done to complete RunMainTask")
|
||||
}
|
||||
if client.calls != 2 {
|
||||
t.Errorf("expected 2 LLM calls, got %d", client.calls)
|
||||
@@ -281,7 +281,7 @@ func TestRunPerFile_ToolCallThenDone(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_ContextCancelled(t *testing.T) {
|
||||
func TestRunMainTask_ContextCancelled(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{taskDoneResponse()}}
|
||||
deps := newTestDeps(client)
|
||||
runner := NewRunner(deps)
|
||||
@@ -290,16 +290,16 @@ func TestRunPerFile_ContextCancelled(t *testing.T) {
|
||||
cancel()
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, _, err := runner.RunPerFile(ctx, msgs, "main.go")
|
||||
completed, _, err := runner.RunMainTask(ctx, msgs, "main.go")
|
||||
if err == nil {
|
||||
t.Error("expected error for cancelled context")
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("cancelled context should not complete RunPerFile")
|
||||
t.Fatal("cancelled context should not complete RunMainTask")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_UnknownTool(t *testing.T) {
|
||||
func TestRunMainTask_UnknownTool(t *testing.T) {
|
||||
content := ""
|
||||
unknownToolResp := &llm.ChatResponse{
|
||||
Choices: []llm.Choice{{
|
||||
@@ -323,19 +323,19 @@ func TestRunPerFile_UnknownTool(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, _, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, _, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if !completed {
|
||||
t.Fatal("expected task_done to complete RunPerFile")
|
||||
t.Fatal("expected task_done to complete RunMainTask")
|
||||
}
|
||||
if client.calls != 2 {
|
||||
t.Errorf("expected 2 calls, got %d", client.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_MaxToolRequestsWithoutTaskDoneDoesNotComplete(t *testing.T) {
|
||||
func TestRunMainTask_MaxToolRequestsWithoutTaskDoneDoesNotComplete(t *testing.T) {
|
||||
content := ""
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{{
|
||||
Choices: []llm.Choice{{Message: llm.ResponseMessage{Content: &content}}},
|
||||
@@ -347,19 +347,19 @@ func TestRunPerFile_MaxToolRequestsWithoutTaskDoneDoesNotComplete(t *testing.T)
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, stop, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, stop, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("RunPerFile completed without task_done")
|
||||
t.Fatal("RunMainTask completed without task_done")
|
||||
}
|
||||
if stop != StopMaxRounds {
|
||||
t.Fatalf("expected StopMaxRounds, got %v", stop)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_EmptyToolResultsStopWithEmptyRounds(t *testing.T) {
|
||||
func TestRunMainTask_EmptyToolResultsStopWithEmptyRounds(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
fileReadToolCallResponse("call_2", `{"path":"main.go"}`),
|
||||
@@ -372,12 +372,12 @@ func TestRunPerFile_EmptyToolResultsStopWithEmptyRounds(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, stop, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, stop, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("RunPerFile completed without task_done")
|
||||
t.Fatal("RunMainTask completed without task_done")
|
||||
}
|
||||
if stop != StopEmptyRounds {
|
||||
t.Fatalf("stop = %v, want StopEmptyRounds", stop)
|
||||
@@ -387,7 +387,7 @@ func TestRunPerFile_EmptyToolResultsStopWithEmptyRounds(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_UncompressibleContextStopsWithCompression(t *testing.T) {
|
||||
func TestRunMainTask_UncompressibleContextStopsWithCompression(t *testing.T) {
|
||||
emptySummary := ""
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
@@ -404,12 +404,12 @@ func TestRunPerFile_UncompressibleContextStopsWithCompression(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", strings.Repeat("word ", 100))}
|
||||
completed, stop, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, stop, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("RunPerFile completed without task_done")
|
||||
t.Fatal("RunMainTask completed without task_done")
|
||||
}
|
||||
if stop != StopCompression {
|
||||
t.Fatalf("stop = %v, want StopCompression", stop)
|
||||
@@ -659,7 +659,7 @@ func graceRoundCommentResponse() *llm.ChatResponse {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_GraceRoundSubmitsComment(t *testing.T) {
|
||||
func TestRunMainTask_GraceRoundSubmitsComment(t *testing.T) {
|
||||
// Round 1: file_read (exhausts budget with MaxToolRequestTimes=1)
|
||||
// Grace round: model calls code_comment
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
@@ -688,9 +688,9 @@ func TestRunPerFile_GraceRoundSubmitsComment(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
completed, stop, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
completed, stop, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if completed {
|
||||
t.Fatal("expected not completed (budget exhausted)")
|
||||
@@ -721,7 +721,7 @@ func TestRunPerFile_GraceRoundSubmitsComment(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_GraceRoundSkippedWhenContextCancelled(t *testing.T) {
|
||||
func TestRunMainTask_GraceRoundSkippedWhenContextCancelled(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
}}
|
||||
@@ -748,7 +748,7 @@ func TestRunPerFile_GraceRoundSkippedWhenContextCancelled(t *testing.T) {
|
||||
runner = NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
_, stop, _ := runner.RunPerFile(ctx, msgs, "main.go")
|
||||
_, stop, _ := runner.RunMainTask(ctx, msgs, "main.go")
|
||||
if stop != StopMaxRounds {
|
||||
t.Fatalf("stop = %v, want StopMaxRounds", stop)
|
||||
}
|
||||
@@ -774,7 +774,7 @@ func (c *cancelAfterNClient) CompletionsWithCtx(ctx context.Context, req llm.Cha
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func TestRunPerFile_GraceRoundNotTriggeredOnEmptyRoundsStop(t *testing.T) {
|
||||
func TestRunMainTask_GraceRoundNotTriggeredOnEmptyRoundsStop(t *testing.T) {
|
||||
client := &fakeClient{responses: []*llm.ChatResponse{
|
||||
fileReadToolCallResponse("call_1", `{"path":"main.go"}`),
|
||||
fileReadToolCallResponse("call_2", `{"path":"main.go"}`),
|
||||
@@ -797,9 +797,9 @@ func TestRunPerFile_GraceRoundNotTriggeredOnEmptyRoundsStop(t *testing.T) {
|
||||
runner := NewRunner(deps)
|
||||
|
||||
msgs := []llm.Message{llm.NewTextMessage("user", "review")}
|
||||
_, stop, err := runner.RunPerFile(context.Background(), msgs, "main.go")
|
||||
_, stop, err := runner.RunMainTask(context.Background(), msgs, "main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if stop != StopEmptyRounds {
|
||||
t.Fatalf("stop = %v, want StopEmptyRounds", stop)
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// Copyright 2026 alibaba/open-code-review Contributors
|
||||
|
||||
// Package llmloop carries the per-file LLM tool-use loop shared by `ocr
|
||||
// review` (diff-based) and `ocr scan` (full-file). It owns the chat
|
||||
// Package llmloop carries the per-subtask MAIN_TASK tool-use loop shared by
|
||||
// `ocr review` (diff-based) and `ocr scan` (full-file). It owns the chat
|
||||
// completion conversation state, three-zone memory compression, tool-call
|
||||
// dispatch (including async comment post-processing), and aggregate token /
|
||||
// warning bookkeeping. Callers above this package render the initial
|
||||
// messages (review uses MAIN_TASK, scan uses FULL_SCAN_TASK) and hand them
|
||||
// in via Runner.RunPerFile.
|
||||
// in via Runner.RunMainTask.
|
||||
//
|
||||
// A subtask is one file in scan and one file group in review, so nothing in
|
||||
// this package may assume its unit of work is a single file.
|
||||
package llmloop
|
||||
|
||||
import (
|
||||
@@ -19,7 +22,7 @@ import (
|
||||
"github.com/alibaba/open-code-review/internal/stdout"
|
||||
)
|
||||
|
||||
// AgentWarning describes a non-fatal warning recorded during a per-file
|
||||
// AgentWarning describes a non-fatal warning recorded during one subtask's
|
||||
// review/scan. The name is kept for backwards compatibility with the
|
||||
// previous internal/agent package.
|
||||
type AgentWarning struct {
|
||||
@@ -41,7 +44,7 @@ type CommentWorkerPool struct {
|
||||
results []model.LlmComment
|
||||
|
||||
// keys tracks per-key WaitGroups so callers can drain only the units
|
||||
// submitted under one key (e.g. one reviewed file) without waiting for
|
||||
// submitted under one key (e.g. one reviewed subtask) without waiting for
|
||||
// — or racing — submissions made under other keys.
|
||||
keysMu sync.Mutex
|
||||
keys map[string]*sync.WaitGroup
|
||||
@@ -69,8 +72,10 @@ func (p *CommentWorkerPool) Submit(f func() ([]model.LlmComment, error)) {
|
||||
//
|
||||
// Callers must guarantee that all SubmitFor calls for a given key
|
||||
// happen-before the matching AwaitKey call for that key — the same contract
|
||||
// as Await, but scoped to one key. Per-file review satisfies this because a
|
||||
// file's tool-use loop finishes submitting before its AwaitKey runs.
|
||||
// as Await, but scoped to one key. Both callers satisfy this because one
|
||||
// MAIN_TASK conversation finishes submitting before the AwaitKey that drains
|
||||
// it. Review reuses one group key across its review rounds, submitting and
|
||||
// draining once per round, so Add never races that key's Wait.
|
||||
func (p *CommentWorkerPool) SubmitFor(key string, f func() ([]model.LlmComment, error)) {
|
||||
p.keysMu.Lock()
|
||||
if p.keys == nil {
|
||||
|
||||
@@ -91,12 +91,12 @@ func wantMeta(t *testing.T, got capturedRequest, want llm.RequestMeta) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_MainTaskIdentity checks that every main_task round carries the
|
||||
// TestRunMainTask_MainTaskIdentity checks that every main_task round carries the
|
||||
// identity of the TaskRecord created for it, including the per-round RequestNo.
|
||||
// An empty provider is covered as its own case: it is the real value for an
|
||||
// unnamed endpoint, so it must still produce identity rather than be read as
|
||||
// "no meta".
|
||||
func TestRunPerFile_MainTaskIdentity(t *testing.T) {
|
||||
func TestRunMainTask_MainTaskIdentity(t *testing.T) {
|
||||
for _, provider := range []string{"openai", ""} {
|
||||
name := provider
|
||||
if name == "" {
|
||||
@@ -115,12 +115,12 @@ func TestRunPerFile_MainTaskIdentity(t *testing.T) {
|
||||
deps.NewRequestMeta = metaFactory(provider, deps.Model)
|
||||
runner := NewRunner(deps)
|
||||
|
||||
if _, _, err := runner.RunPerFile(
|
||||
if _, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
); err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
|
||||
reqs := client.requests()
|
||||
@@ -152,9 +152,9 @@ func TestRunPerFile_MainTaskIdentity(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_GraceRoundIsAMainTaskRound verifies that the grace round uses
|
||||
// TestRunMainTask_GraceRoundIsAMainTaskRound verifies that the grace round uses
|
||||
// the next main_task identity and records both responses and errors.
|
||||
func TestRunPerFile_GraceRoundIsAMainTaskRound(t *testing.T) {
|
||||
func TestRunMainTask_GraceRoundIsAMainTaskRound(t *testing.T) {
|
||||
// Delay the grace response so duration assertions are stable across platforms.
|
||||
const graceDelay = 20 * time.Millisecond
|
||||
|
||||
@@ -206,13 +206,13 @@ func TestRunPerFile_GraceRoundIsAMainTaskRound(t *testing.T) {
|
||||
}
|
||||
deps.NewRequestMeta = metaFactory("openai", deps.Model)
|
||||
|
||||
completed, stop, err := NewRunner(deps).RunPerFile(
|
||||
completed, stop, err := NewRunner(deps).RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
if completed || stop != StopMaxRounds {
|
||||
t.Fatalf("completed = %v, stop = %v; want false, StopMaxRounds", completed, stop)
|
||||
@@ -243,18 +243,18 @@ func TestRunPerFile_GraceRoundIsAMainTaskRound(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunPerFile_NoIdentityWhenFactoryNil is the scan guarantee: the Runner is
|
||||
// TestRunMainTask_NoIdentityWhenFactoryNil is the scan guarantee: the Runner is
|
||||
// shared, and with NewRequestMeta left nil no request may carry identity.
|
||||
func TestRunPerFile_NoIdentityWhenFactoryNil(t *testing.T) {
|
||||
func TestRunMainTask_NoIdentityWhenFactoryNil(t *testing.T) {
|
||||
client := &metaCaptureClient{respond: func(int) *llm.ChatResponse { return taskDoneResponse() }}
|
||||
runner := NewRunner(newTestDeps(client)) // NewRequestMeta unset, as scan leaves it
|
||||
|
||||
if _, _, err := runner.RunPerFile(
|
||||
if _, _, err := runner.RunMainTask(
|
||||
context.Background(),
|
||||
[]llm.Message{llm.NewTextMessage("user", "review this file")},
|
||||
"main.go",
|
||||
); err != nil {
|
||||
t.Fatalf("RunPerFile: %v", err)
|
||||
t.Fatalf("RunMainTask: %v", err)
|
||||
}
|
||||
|
||||
reqs := client.requests()
|
||||
@@ -345,9 +345,10 @@ func TestRunCompression_NoIdentityWhenFactoryNil(t *testing.T) {
|
||||
|
||||
// TestReLocation_Identity pins the field that is easy to get wrong: FilePath is
|
||||
// the comment's path, which is the file session the re-location record was
|
||||
// written to. It normally equals the tool loop's newPath, because executeToolCall
|
||||
// overrides the path argument with it — so the test leaves newPath empty, the one
|
||||
// case where the argument survives, to show which of the two identity follows.
|
||||
// written to — not the tool loop's taskKey, which is only the fallback for a
|
||||
// comment that named no path. The test passes an empty taskKey against a
|
||||
// comment that does name one, so a regression that switched the identity to
|
||||
// taskKey would show up as an empty FilePath rather than as other.go.
|
||||
func TestReLocation_Identity(t *testing.T) {
|
||||
collector := tool.NewCommentCollector()
|
||||
reg := tool.NewRegistry()
|
||||
@@ -380,8 +381,8 @@ func TestReLocation_Identity(t *testing.T) {
|
||||
NewRequestMeta: metaFactory("openai", "fake"),
|
||||
})
|
||||
|
||||
// newPath is empty so the path override does not fire and the comment keeps
|
||||
// the target the tool call named, other.go.
|
||||
// taskKey is empty, so the fallback cannot supply a path and the comment
|
||||
// keeps the target the tool call named, other.go.
|
||||
cp := r.executeToolCall(context.Background(), "", llm.ToolCall{
|
||||
Function: llm.FunctionCall{
|
||||
Name: tool.CodeComment.Name(),
|
||||
|
||||
@@ -50,7 +50,7 @@ func (g *gatedLLMClient) CompletionsWithCtx(ctx context.Context, _ llm.ChatReque
|
||||
|
||||
// concurrentFakeClient is goroutine-safe and distinguishes compression
|
||||
// requests (no tools attached) from main-loop requests (tools attached),
|
||||
// mirroring how runCompression and RunPerFile build their ChatRequests.
|
||||
// mirroring how runCompression and RunMainTask build their ChatRequests.
|
||||
type concurrentFakeClient struct {
|
||||
compressionCalls atomic.Int64
|
||||
}
|
||||
@@ -690,7 +690,7 @@ func TestAddNextMessage_NoStartThenCancelSameCall(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPerFile_ConcurrentFilesCompression_Race(t *testing.T) {
|
||||
func TestRunMainTask_ConcurrentFilesCompression_Race(t *testing.T) {
|
||||
t_tempDir = t.TempDir()
|
||||
tpl := template.Template{
|
||||
MemoryCompressionTask: template.LlmConversation{
|
||||
@@ -710,7 +710,7 @@ func TestRunPerFile_ConcurrentFilesCompression_Race(t *testing.T) {
|
||||
CommentCollector: tool.NewCommentCollector(),
|
||||
// MainToolDefs must be non-empty: the fake client classifies a
|
||||
// request with no tools as a compression request, mirroring how
|
||||
// RunPerFile and runCompression build their ChatRequests.
|
||||
// RunMainTask and runCompression build their ChatRequests.
|
||||
MainToolDefs: []llm.ToolDef{{Type: "function", Function: llm.FunctionDef{Name: "file_read"}}},
|
||||
Session: session.New(t_tempDir, "main", "test-model", session.SessionOptions{ReviewMode: "diff"}),
|
||||
})
|
||||
@@ -725,14 +725,14 @@ func TestRunPerFile_ConcurrentFilesCompression_Race(t *testing.T) {
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
msgs := []llm.Message{msg("system", "sys"), msg("user", "review this file")}
|
||||
_, _, err := r.RunPerFile(context.Background(), msgs, fmt.Sprintf("f%d.go", i))
|
||||
_, _, err := r.RunMainTask(context.Background(), msgs, fmt.Sprintf("f%d.go", i))
|
||||
errs[i] = err
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
for i, err := range errs {
|
||||
if err != nil {
|
||||
t.Errorf("file %d: RunPerFile: %v", i, err)
|
||||
t.Errorf("file %d: RunMainTask: %v", i, err)
|
||||
}
|
||||
}
|
||||
// Guard against the test going vacuous: if no compression request was
|
||||
|
||||
@@ -790,7 +790,7 @@ func (a *Agent) executeSubtask(ctx context.Context, it model.ScanItem) (bool, st
|
||||
return false, "", nil
|
||||
}
|
||||
|
||||
completed, stop, err := a.runner.RunPerFile(ctx, messages, it.Path)
|
||||
completed, stop, err := a.runner.RunMainTask(ctx, messages, it.Path)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
|
||||
@@ -497,7 +497,7 @@ func TestScanAgent_WaitBackground_NoLeakOnRun(t *testing.T) {
|
||||
|
||||
// Verify that a.Run has NOT returned yet (held by a.runner.WaitBackground()).
|
||||
// Use a timeout rather than default: the main goroutine's entire post-loop
|
||||
// cleanup (RunPerFile return → dispatchSubtasks → Finalize) completes in
|
||||
// cleanup (RunMainTask return → dispatchSubtasks → Finalize) completes in
|
||||
// under 1ms on any machine, so 200ms is a generous upper bound. If Run
|
||||
// returns within this window, WaitBackground is not holding it.
|
||||
select {
|
||||
|
||||
@@ -125,7 +125,7 @@ func TestScanRequestsCarryNoIdentity(t *testing.T) {
|
||||
client := &identityProbeClient{reply: "no findings"}
|
||||
a := newProbeAgent(t, makeTemplateWithFullScan(), client, nil)
|
||||
|
||||
// executeSubtask drives llmloop.RunPerFile, the code path scan shares
|
||||
// executeSubtask drives llmloop.RunMainTask, the code path scan shares
|
||||
// with review — so this is the assertion that NewAgent leaves
|
||||
// Deps.NewRequestMeta nil.
|
||||
if _, _, err := a.executeSubtask(context.Background(), model.ScanItem{
|
||||
|
||||
Reference in New Issue
Block a user