This repository was archived by the owner on Jun 2, 2026. It is now read-only.
feat: Improve Flow task description, message, and progress report#570
Merged
Conversation
Contributor
Summary by CodeRabbit
WalkthroughAdds structured task progress reporting end-to-end: DB migration and Task.Report, versioned report model and Tracker, workflow snapshotting and best-effort persistence, Temporal report activity and store API, message/operation helpers, and consolidated test helpers. ChangesTask Progress Reporting Implementation
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
Comment |
🔐 TruffleHog Secret Scan✅ No secrets or credentials found! Your code has been scanned for 700+ types of secrets and credentials. All clear! 🎉 🕐 Last updated: 2026-05-22 06:50:14 UTC | Commit: 7ed7eb3 |
Test Results9 779 tests +14 9 779 ✅ +14 7m 2s ⏱️ -27s Results for commit a4e010d. ± Comparison against base commit 9b230cd. This pull request removes 1 and adds 15 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (3)
flow/internal/task/manager/manager.go (1)
589-593: ⚡ Quick winRefactor inline assignments in conditionals for guideline compliance.
Lines 589 and 592 use inline assign-and-condition; split both into separate assignment and
ifstatements.As per coding guidelines: "Split assign-and-condition into two statements; prefer separate `derr := action()` and `if derr != nil` over `if derr := action(); derr != nil`."Proposed change
- if desc := strings.TrimSpace(req.Description); desc != "" { + desc := strings.TrimSpace(req.Description) + if desc != "" { return desc } - if summary, err := operations.SummaryFromWrapper(req.Operation); err == nil { + summary, derr := operations.SummaryFromWrapper(req.Operation) + if derr == nil { return summary }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/manager/manager.go` around lines 589 - 593, The code uses inline assign-and-condition patterns for Description and Summary extraction; change the two occurrences to first assign the result to local variables (e.g., desc := strings.TrimSpace(req.Description) and summary, err := operations.SummaryFromWrapper(req.Operation)) and then use separate if statements to check the values/err (if desc != "" { return desc } and if err == nil { return summary }), updating the existing branches around req.Description and operations.SummaryFromWrapper to follow the split-assignment guideline.flow/internal/task/operations/summary.go (1)
63-65: ⚡ Quick winSplit inline error assignment to satisfy repo Go style.
Line 63 uses inline assign-and-condition. Please split it into two statements to match the enforced repository convention.
As per coding guidelines: "Split assign-and-condition into two statements; prefer separate `derr := action()` and `if derr != nil` over `if derr := action(); derr != nil`."Proposed change
- if len(w.Info) > 0 { - if err := json.Unmarshal(w.Info, &info); err != nil { - return "", fmt.Errorf("unmarshal firmware info: %w", err) - } - } + if len(w.Info) > 0 { + derr := json.Unmarshal(w.Info, &info) + if derr != nil { + return "", fmt.Errorf("unmarshal firmware info: %w", derr) + } + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/operations/summary.go` around lines 63 - 65, The inline assign-and-check using json.Unmarshal in summary.go should be split into two statements to follow repo Go style: call derr := json.Unmarshal(w.Info, &info) on its own line, then check if derr != nil and return fmt.Errorf("unmarshal firmware info: %w", derr); keep the same error message and use the new derr variable in the fmt.Errorf call so behavior remains identical.openapi/spec.yaml (1)
17337-17340: ⚡ Quick winDefine a typed schema for
progressReportinstead of a free-form object.
additionalProperties: truemakes this contract opaque for generated clients and weakens validation. Please reference a dedicatedTaskProgressReportschema aligned with the protobuf/domain progress model.Proposed OpenAPI change
- progressReport: - type: object - description: Structured progress by execution stage. Omitted until the task produces progress data. - additionalProperties: true + progressReport: + $ref: '`#/components/schemas/TaskProgressReport`'As per coding guidelines:
openapi/spec.yaml: “Review the OpenAPI specification, check for consistency and correctness, check for misspellings and grammatical errors.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openapi/spec.yaml` around lines 17337 - 17340, The current progressReport schema is an untyped free-form object; replace it with a typed reference to a new components schema (e.g., TaskProgressReport) and remove additionalProperties:true to restore strict validation. Add a components.schemas.TaskProgressReport that mirrors the protobuf/domain progress model (fields such as stage/status, percentage/progress, details/message, startedAt, finishedAt, and optional metadata/map with a clear value type), then update the progressReport property to use $ref: '`#/components/schemas/TaskProgressReport`' and adjust any examples/responses that reference progressReport to match the new structure.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@api/pkg/api/model/task.go`:
- Around line 64-66: In FromProto, ensure t.ProgressReport is cleared before
conditionally assigning to avoid leaking stale data: first assign raw :=
task.GetProgressReport(), set t.ProgressReport = nil, then if raw != "" &&
json.Valid([]byte(raw)) set t.ProgressReport = json.RawMessage(raw); update
references to task.GetProgressReport and t.ProgressReport accordingly to follow
the split assign-and-condition style and preserve the json.Valid check.
In `@flow/internal/converter/protobuf/converter.go`:
- Line 469: The change mutates Task.operation semantics in-place; instead add a
new stable label field (e.g., Task.operation_type or Task.operation_label) in
the protobuf and keep the existing Task.operation value for backward
compatibility, then in converter.go set the new field using
operations.OperationTypeFromWrapper(task.Operation) while leaving task.Operation
unchanged; update the protobuf message (add the new field with a new tag and
deprecation comment for operation if desired), adjust any marshalling/usage
sites to read the new field where a stable enum/label is required, and ensure
converters/serializers reference the new field name rather than repurposing
Task.operation.
In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go`:
- Around line 392-399: The assign-and-condition pattern used with
updateTaskProgress should be split into two statements: first call
updateTaskProgress and assign its result to a named error variable (e.g., uerr)
and then use a separate if uerr != nil check to return; locate the call to
updateTaskProgress that currently reads if uerr := updateTaskProgress(...); uerr
!= nil and replace it with a standalone assignment to uerr (preserving the same
arguments: ctx, taskID, progress.StageMessage(stageIndex, totalStages,
componentType), prog) followed by if uerr != nil { return prog, uerr }.
- Around line 382-391: The code currently sets componentType from only
stage.Steps[0], which misreports stages that run multiple component types;
update the logic that builds prog.Stages so it inspects all entries in
stage.Steps: if every step has the same ComponentType use
devicetypes.ComponentTypeToString(thatType), otherwise set ComponentType to a
deterministic marker like "mixed" (or an existing devicetypes constant if
available). Change the block around stage.Steps, the calculation of
componentType, and the prog.Stages append so progress.Stage.ComponentType
reflects either the unanimous type or "mixed".
- Around line 112-116: The current code discards the original operation error
(err) when report.Marshal() fails by returning only merr; update the return to
preserve and surface both errors so callers see the original failure and the
marshal failure. In the block around report.Marshal() (variables raw, merr and
the call report.Marshal()), change the error returned to wrap the original err
and include merr (for example by using fmt.Errorf to wrap err and
annotate/include merr), so the final error message contains the original
operation error plus the marshaling error details.
In
`@flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go`:
- Around line 154-155: The test setup calls expectTaskUpdateActivities(env)
twice in a row, causing redundant mock wiring; remove the duplicate calls so
each test invokes expectTaskUpdateActivities(env) only once (e.g., leave a
single expectTaskUpdateActivities(env) where currently there are back-to-back
calls and do the same for the other duplicated pair), ensuring no other setup
relies on the second invocation.
In `@flow/internal/task/progress/report.go`:
- Around line 90-92: Split the assign-and-condition into two statements: first
compute idx with idx := strings.IndexByte(msg, '\n') and then use if idx >= 0 {
msg = strings.TrimSpace(msg[:idx]) } so the logic around trimming msg on the
first newline remains identical; update the block that currently reads if idx :=
strings.IndexByte(msg, '\n'); idx >= 0 { ... } accordingly (the variables msg
and idx and the strings.IndexByte call identify the location).
In `@flow/internal/task/store/postgres.go`:
- Around line 164-166: The UpdateTaskStatus calls are using s.pg.DB directly
(bypassing transaction propagation); change both calls that pass s.pg.DB into
taskDao.UpdateTaskStatus to use the transaction-aware database helper s.idb(ctx)
instead so RunInTransaction context is respected—i.e., locate the
UpdateTaskStatus invocations and replace the s.pg.DB argument with s.idb(ctx)
(ensure ctx is passed through and s.idb(ctx) returns the same DB interface
expected by UpdateTaskStatus).
---
Nitpick comments:
In `@flow/internal/task/manager/manager.go`:
- Around line 589-593: The code uses inline assign-and-condition patterns for
Description and Summary extraction; change the two occurrences to first assign
the result to local variables (e.g., desc := strings.TrimSpace(req.Description)
and summary, err := operations.SummaryFromWrapper(req.Operation)) and then use
separate if statements to check the values/err (if desc != "" { return desc }
and if err == nil { return summary }), updating the existing branches around
req.Description and operations.SummaryFromWrapper to follow the split-assignment
guideline.
In `@flow/internal/task/operations/summary.go`:
- Around line 63-65: The inline assign-and-check using json.Unmarshal in
summary.go should be split into two statements to follow repo Go style: call
derr := json.Unmarshal(w.Info, &info) on its own line, then check if derr != nil
and return fmt.Errorf("unmarshal firmware info: %w", derr); keep the same error
message and use the new derr variable in the fmt.Errorf call so behavior remains
identical.
In `@openapi/spec.yaml`:
- Around line 17337-17340: The current progressReport schema is an untyped
free-form object; replace it with a typed reference to a new components schema
(e.g., TaskProgressReport) and remove additionalProperties:true to restore
strict validation. Add a components.schemas.TaskProgressReport that mirrors the
protobuf/domain progress model (fields such as stage/status,
percentage/progress, details/message, startedAt, finishedAt, and optional
metadata/map with a clear value type), then update the progressReport property
to use $ref: '`#/components/schemas/TaskProgressReport`' and adjust any
examples/responses that reference progressReport to match the new structure.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: de3c7160-2f5f-40f3-9e2e-afc68a09ed80
⛔ Files ignored due to path filters (2)
flow/pkg/proto/v1/flow.pb.gois excluded by!**/*.pb.go,!**/*.pb.goworkflow-schema/flow/protobuf/v1/flow.pb.gois excluded by!**/*.pb.go,!**/*.pb.go
📒 Files selected for processing (39)
api/pkg/api/model/task.goflow/internal/converter/dao/converter.goflow/internal/converter/protobuf/converter.goflow/internal/db/migrations/20260521100000_add_task_progress_report.down.sqlflow/internal/db/migrations/20260521100000_add_task_progress_report.up.sqlflow/internal/db/model/task.goflow/internal/scheduler/taskschedule/dispatcher_test.goflow/internal/scheduler/taskschedule/template.goflow/internal/scheduler/taskschedule/template_test.goflow/internal/task/conflict/store_mock_test.goflow/internal/task/executor/temporalworkflow/activity/activity.goflow/internal/task/executor/temporalworkflow/activity/registry.goflow/internal/task/executor/temporalworkflow/activity/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/bringup.goflow/internal/task/executor/temporalworkflow/workflow/bringup_test.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/helpers.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.goflow/internal/task/manager/manager.goflow/internal/task/manager/notifying_store.goflow/internal/task/operationrules/rules.goflow/internal/task/operations/summary.goflow/internal/task/operations/summary_test.goflow/internal/task/progress/report.goflow/internal/task/progress/report_test.goflow/internal/task/store/postgres.goflow/internal/task/store/store.goflow/internal/task/task/task.goflow/proto/v1/flow.protoopenapi/spec.yamlworkflow-schema/flow/proto/v1/flow.proto
Comment on lines
+64
to
+66
| if raw := task.GetProgressReport(); raw != "" && json.Valid([]byte(raw)) { | ||
| t.ProgressReport = json.RawMessage(raw) | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the file and print relevant sections around the reported lines
FILE="api/pkg/api/model/task.go"
echo "=== File: $FILE ==="
ls -la "$FILE"
echo "--- Lines 1-140 ---"
nl -ba "$FILE" | sed -n '1,140p'
echo "--- Lines 40-110 ---"
nl -ba "$FILE" | sed -n '40,110p'
echo "--- Lines 110-220 ---"
nl -ba "$FILE" | sed -n '110,220p'
# Find FromProto definition and any uses of APIRackTask.FromProto
echo "=== Search: FromProto ==="
rg -n "func \\(.*\\) FromProto\\b" api/pkg/api/model -S
rg -n "FromProto\\(" -S api/pkg/api/model api/pkg | head -n 200
# Search for APIRackTask reuse patterns (assign to variable then call FromProto multiple times)
echo "=== Search: APIRackTask usage ==="
rg -n "\\bAPIRackTask\\b" api/pkg -S | head -n 200Repository: NVIDIA/infra-controller-rest
Length of output: 268
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="api/pkg/api/model/task.go"
echo "=== File: $FILE ==="
ls -la "$FILE"
echo "--- Lines 1-140 (numbered via awk) ---"
awk 'NR>=1 && NR<=140 {printf "%5d\t%s\n", NR, $0}' "$FILE"
echo "--- Lines 50-110 (target area) ---"
awk 'NR>=50 && NR<=110 {printf "%5d\t%s\n", NR, $0}' "$FILE"
echo "=== FromProto definitions ==="
rg -n --hidden --no-messages "func \\(.*\\) FromProto\\b" api/pkg/api/model/task.go api/pkg/api/model -S || true
echo "=== Call sites of FromProto ==="
rg -n --hidden --no-messages "\\.FromProto\\(" api/pkg -S | head -n 200 || true
echo "=== APIRackTask struct and ProgressReport fields ==="
rg -n --hidden --no-messages "type APIRackTask|ProgressReport" api/pkg/api/model/task.go -S || trueRepository: NVIDIA/infra-controller-rest
Length of output: 11627
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== Search: NewAPIRackTask usages ==="
rg -n --hidden --no-messages "NewAPIRackTask\\(" api/pkg -S || true
echo "=== Search: APIRackTask instantiations ==="
rg -n --hidden --no-messages "\\bAPIRackTask\\b" api/pkg -S | head -n 200 || true
echo "=== Search: direct FromProto calls on APIRackTask variable ==="
rg -n --hidden --no-messages "\\.FromProto\\(.*\\)" api/pkg/api -S | rg -n "task\\.go:81|APIRackTask|rack task" || true
# More targeted: find "APIRackTask" followed nearby by ".FromProto("
rg -n --hidden --no-messages "APIRackTask[\\s\\S]{0,200}\\.FromProto\\(" api/pkg -S || true
# Find task model usage in handlers
echo "=== Search: flowv1.Task conversions (NewAPIRackTask or FromProto) ==="
rg -n --hidden --no-messages "flowv1\\.Task" api/pkg -S || trueRepository: NVIDIA/infra-controller-rest
Length of output: 5979
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Helper to print specific ranges with awk
print_range () {
local file="$1"; local start="$2"; local end="$3"
echo "=== $file : lines $start-$end ==="
awk -v s="$start" -v e="$end" 'NR>=s && NR<=e {printf "%5d\t%s\n", NR, $0}' "$file"
}
print_range "api/pkg/api/handler/task.go" 150 260
print_range "api/pkg/api/handler/task.go" 340 430
print_range "api/pkg/api/handler/task.go" 520 610
print_range "api/pkg/api/handler/task.go" 690 780
print_range "api/pkg/api/model/task_test.go" 1 120
print_range "api/pkg/api/model/task_test.go" 120 240
# Reuse patterns were hinted at in handler/task_test.go
print_range "api/pkg/api/handler/task_test.go" 160 260
print_range "api/pkg/api/handler/task_test.go" 260 360
print_range "api/pkg/api/handler/task_test.go" 560 660
echo "=== Search: ProgressReport occurrences in tests ==="
rg -n --hidden --no-messages "ProgressReport" api/pkg/api/model api/pkg/api/handler -S || true
echo "=== Search: FromProto calls on non-pointer APIRackTask variables ==="
rg -n --hidden --no-messages "var\\s+apiTask\\s+model\\.APIRackTask|\\[\\]model\\.APIRackTask|\\.FromProto\\(.*\\)" api/pkg/api/handler/task_test.go -S || trueRepository: NVIDIA/infra-controller-rest
Length of output: 34831
Fix FromProto ProgressReport assignment (stale field + style rule)
FromProtoonly setsProgressReportwhenprogressReportis non-empty and valid JSON, but it never clears any existing value; if the receiver is ever reused, stale data can leak.- The
if raw := ...; ...pattern violates the repo’s “split assign-and-condition” Go style.
Proposed change
- if raw := task.GetProgressReport(); raw != "" && json.Valid([]byte(raw)) {
- t.ProgressReport = json.RawMessage(raw)
- }
+ raw := task.GetProgressReport()
+ t.ProgressReport = nil
+ if raw != "" && json.Valid([]byte(raw)) {
+ t.ProgressReport = json.RawMessage(raw)
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if raw := task.GetProgressReport(); raw != "" && json.Valid([]byte(raw)) { | |
| t.ProgressReport = json.RawMessage(raw) | |
| } | |
| raw := task.GetProgressReport() | |
| t.ProgressReport = nil | |
| if raw != "" && json.Valid([]byte(raw)) { | |
| t.ProgressReport = json.RawMessage(raw) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@api/pkg/api/model/task.go` around lines 64 - 66, In FromProto, ensure
t.ProgressReport is cleared before conditionally assigning to avoid leaking
stale data: first assign raw := task.GetProgressReport(), set t.ProgressReport =
nil, then if raw != "" && json.Valid([]byte(raw)) set t.ProgressReport =
json.RawMessage(raw); update references to task.GetProgressReport and
t.ProgressReport accordingly to follow the split assign-and-condition style and
preserve the json.Valid check.
| pbTask := &pb.Task{ | ||
| Id: UUIDTo(task.ID), | ||
| Operation: opStr, | ||
| Operation: operations.OperationTypeFromWrapper(task.Operation), |
Contributor
There was a problem hiding this comment.
Do not repurpose existing Task.operation semantics in-place.
Line 469 changes Task.operation from a human-readable operation value to a stable type label on the same field, which is a backward-incompatible API contract shift for existing clients. Add a new field for the stable label and keep current operation semantics through a deprecation window.
As per coding guidelines: **/*.proto: Review the Protobuf definitions, point out issues relative to compatibility, and expressiveness.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flow/internal/converter/protobuf/converter.go` at line 469, The change
mutates Task.operation semantics in-place; instead add a new stable label field
(e.g., Task.operation_type or Task.operation_label) in the protobuf and keep the
existing Task.operation value for backward compatibility, then in converter.go
set the new field using operations.OperationTypeFromWrapper(task.Operation)
while leaving task.Operation unchanged; update the protobuf message (add the new
field with a new tag and deprecation comment for operation if desired), adjust
any marshalling/usage sites to read the new field where a stable enum/label is
required, and ensure converters/serializers reference the new field name rather
than repurposing Task.operation.
Comment on lines
+382
to
+391
| componentType := "" | ||
| if len(stage.Steps) > 0 { | ||
| componentType = devicetypes.ComponentTypeToString(stage.Steps[0].ComponentType) | ||
| } | ||
|
|
||
| prog.Stages = append(prog.Stages, progress.Stage{ | ||
| Number: stage.Number, | ||
| Status: progress.StageStatusRunning, | ||
| ComponentType: componentType, | ||
| }) |
Contributor
There was a problem hiding this comment.
Avoid misreporting component type for mixed-component stages.
componentType is derived from only the first step, but this stage model executes multiple component types in parallel. The progress payload can therefore report the wrong type.
💡 Proposed fix
componentType := ""
if len(stage.Steps) > 0 {
componentType = devicetypes.ComponentTypeToString(stage.Steps[0].ComponentType)
+ for _, s := range stage.Steps[1:] {
+ if s.ComponentType != stage.Steps[0].ComponentType {
+ componentType = "MIXED"
+ break
+ }
+ }
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go` around
lines 382 - 391, The code currently sets componentType from only stage.Steps[0],
which misreports stages that run multiple component types; update the logic that
builds prog.Stages so it inspects all entries in stage.Steps: if every step has
the same ComponentType use devicetypes.ComponentTypeToString(thatType),
otherwise set ComponentType to a deterministic marker like "mixed" (or an
existing devicetypes constant if available). Change the block around
stage.Steps, the calculation of componentType, and the prog.Stages append so
progress.Stage.ComponentType reflects either the unanimous type or "mixed".
Comment on lines
+392
to
+399
| if uerr := updateTaskProgress( | ||
| ctx, | ||
| taskID, | ||
| progress.StageMessage(stageIndex, totalStages, componentType), | ||
| prog, | ||
| ); uerr != nil { | ||
| return prog, uerr | ||
| } |
Contributor
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Split assignment and condition in error handling.
Please split this into two statements per repository convention.
💡 Proposed fix
- if uerr := updateTaskProgress(
- ctx,
- taskID,
- progress.StageMessage(stageIndex, totalStages, componentType),
- prog,
- ); uerr != nil {
+ uerr := updateTaskProgress(
+ ctx,
+ taskID,
+ progress.StageMessage(stageIndex, totalStages, componentType),
+ prog,
+ )
+ if uerr != nil {
return prog, uerr
}As per coding guidelines: Split assign-and-condition into two statements; prefer separate derr := action() and if derr != nil over if derr := action(); derr != nil.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if uerr := updateTaskProgress( | |
| ctx, | |
| taskID, | |
| progress.StageMessage(stageIndex, totalStages, componentType), | |
| prog, | |
| ); uerr != nil { | |
| return prog, uerr | |
| } | |
| uerr := updateTaskProgress( | |
| ctx, | |
| taskID, | |
| progress.StageMessage(stageIndex, totalStages, componentType), | |
| prog, | |
| ) | |
| if uerr != nil { | |
| return prog, uerr | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go` around
lines 392 - 399, The assign-and-condition pattern used with updateTaskProgress
should be split into two statements: first call updateTaskProgress and assign
its result to a named error variable (e.g., uerr) and then use a separate if
uerr != nil check to return; locate the call to updateTaskProgress that
currently reads if uerr := updateTaskProgress(...); uerr != nil and replace it
with a standalone assignment to uerr (preserving the same arguments: ctx,
taskID, progress.StageMessage(stageIndex, totalStages, componentType), prog)
followed by if uerr != nil { return prog, uerr }.
Comment on lines
+90
to
+92
| if idx := strings.IndexByte(msg, '\n'); idx >= 0 { | ||
| msg = strings.TrimSpace(msg[:idx]) | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="flow/internal/task/progress/report.go"
# Show surrounding context with line numbers
nl -ba "$file" | sed -n '70,110p'Repository: NVIDIA/infra-controller-rest
Length of output: 115
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="flow/internal/task/progress/report.go"
awk 'NR>=60 && NR<=110 {printf "%4d\t%s\n", NR, $0}' "$file"Repository: NVIDIA/infra-controller-rest
Length of output: 1485
Split the newline index assignment and condition into two statements.
The code uses assign-and-condition in one if (if idx := ...; idx >= 0), which conflicts with the repository’s Go style guideline. Split it for lint/style compliance.
Proposed change
- if idx := strings.IndexByte(msg, '\n'); idx >= 0 {
+ idx := strings.IndexByte(msg, '\n')
+ if idx >= 0 {
msg = strings.TrimSpace(msg[:idx])
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if idx := strings.IndexByte(msg, '\n'); idx >= 0 { | |
| msg = strings.TrimSpace(msg[:idx]) | |
| } | |
| idx := strings.IndexByte(msg, '\n') | |
| if idx >= 0 { | |
| msg = strings.TrimSpace(msg[:idx]) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flow/internal/task/progress/report.go` around lines 90 - 92, Split the
assign-and-condition into two statements: first compute idx with idx :=
strings.IndexByte(msg, '\n') and then use if idx >= 0 { msg =
strings.TrimSpace(msg[:idx]) } so the logic around trimming msg on the first
newline remains identical; update the block that currently reads if idx :=
strings.IndexByte(msg, '\n'); idx >= 0 { ... } accordingly (the variables msg
and idx and the strings.IndexByte call identify the location).
kunzhao-nv
force-pushed
the
feat/flow-better-info
branch
2 times, most recently
from
39bde08 to
fd89798
Compare
kunzhao-nv
force-pushed
the
feat/flow-better-info
branch
from
fd89798 to
4366cfa
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (4)
flow/internal/task/store/postgres.go (1)
167-175:⚠️ Potential issue | 🟠 Major | ⚡ Quick winUse tx-aware DB handle in report read/write paths.
Line 167, Line 175, Line 188, and Line 199 use
s.pg.DBdirectly, which can bypassRunInTransactioncontext propagation and break atomicity for callers expecting one transaction boundary.Suggested fix
- existing, gerr := model.GetTask(ctx, s.pg.DB, arg.ID) + existing, gerr := model.GetTask(ctx, s.idb(ctx), arg.ID) @@ - err := taskDao.UpdateTaskStatus(ctx, s.pg.DB, arg.Status, arg.Message, merged) + err := taskDao.UpdateTaskStatus(ctx, s.idb(ctx), arg.Status, arg.Message, merged) @@ - existing, err := model.GetTask(ctx, s.pg.DB, arg.ID) + existing, err := model.GetTask(ctx, s.idb(ctx), arg.ID) @@ - if uerr := taskDao.UpdateTaskReport(ctx, s.pg.DB, merged); uerr != nil { + if uerr := taskDao.UpdateTaskReport(ctx, s.idb(ctx), merged); uerr != nil { return errors.GRPCErrorInternal(uerr.Error()) }Also applies to: 188-200
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/store/postgres.go` around lines 167 - 175, The code is using the global handle s.pg.DB when reading/updating reports which can bypass a RunInTransaction context; change the model.GetTask, reportpkg.Merge read path and taskDao.UpdateTaskStatus call to use the transaction-aware DB handle derived from the incoming ctx instead of s.pg.DB (e.g., obtain the tx/DB from ctx or the pg wrapper used elsewhere in this package and pass that handle into model.GetTask and taskDao.UpdateTaskStatus) so all operations (GetTask, Merge, UpdateTaskStatus) participate in the caller's transaction boundary.flow/internal/task/executor/temporalworkflow/workflow/helpers.go (3)
96-98: 🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick winSplit assignment and condition for activity error handling.
Line 96 should use a separate assignment and
ifcheck to comply with the repository rule.As per coding guidelines: `Split assign-and-condition into two statements; prefer separate derr := action() and if derr != nil over if derr := action(); derr != nil`.Suggested fix
- if err := workflow.ExecuteActivity(actx, activity.NameUpdateTaskReport, arg).Get(ctx, nil); err != nil { + err = workflow.ExecuteActivity(actx, activity.NameUpdateTaskReport, arg).Get(ctx, nil) + if err != nil { log.Warn().Err(err).Str("task_id", taskID.String()).Msg("failed to update task report") }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go` around lines 96 - 98, The current inline assignment in the activity call should be split into two statements: first call workflow.ExecuteActivity(actx, activity.NameUpdateTaskReport, arg).Get(ctx, nil) and assign its error to a variable (e.g., derr), then use a separate if derr != nil { ... } block to handle logging; update the code that references taskID and log.Warn().Err(...) to use that derr variable (preserve the same log message and fields) so it conforms to the repository rule against assign-and-condition.
120-124:⚠️ Potential issue | 🟠 Major | ⚡ Quick winPreserve the original operation error when report marshal fails.
At Line 123, returning only the marshal error can hide the original operation failure (
err), which breaks the function’s terminal-error semantics.Suggested fix
var merr error raw, merr = rep.MarshalRaw() if merr != nil { - return fmt.Errorf("marshal task report: %w", merr) + if err != nil { + return errors.Join(err, fmt.Errorf("marshal task report: %w", merr)) + } + return fmt.Errorf("marshal task report: %w", merr) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go` around lines 120 - 124, When rep.MarshalRaw() fails we currently return only merr which loses the original operation error stored in err; modify the error returned in that branch so the original err is preserved as the primary wrapped error and the marshal failure is added as context. Concretely, in the block that calls rep.MarshalRaw() (variables err, merr, raw, and call rep.MarshalRaw()), return a combined error that wraps err with fmt.Errorf and includes the marshal error message (e.g., wrap err with %w and append the marshal error text) so callers retain the original terminal error while still seeing marshal failure details.
368-371:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winAvoid misreporting component type for mixed-component stages.
Line 370 derives
componentTypefrom only the first step; stages with mixed component types will produce misleading progress data.Suggested fix
componentType := "" if len(stage.Steps) > 0 { componentType = devicetypes.ComponentTypeToString(stage.Steps[0].ComponentType) + for _, s := range stage.Steps[1:] { + if s.ComponentType != stage.Steps[0].ComponentType { + componentType = "mixed" + break + } + } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go` around lines 368 - 371, The current logic sets componentType from only stage.Steps[0], which misreports when a stage has mixed component types; update the computation in the helper where componentType is set to iterate all step.ComponentType values (from stage.Steps) and determine if they are all identical—only call devicetypes.ComponentTypeToString when uniform; otherwise set componentType to a clear sentinel (e.g., "mixed" or empty) to indicate multiple component types so progress reporting is accurate.
🧹 Nitpick comments (3)
flow/internal/task/message/message.go (1)
58-60: ⚡ Quick winSplit the inline
ifassignment into two statements.Please avoid scoped assignment in the
ifcondition here to match repo Go style.Proposed change
- if idx := strings.IndexByte(msg, '\n'); idx >= 0 { + idx := strings.IndexByte(msg, '\n') + if idx >= 0 { msg = strings.TrimSpace(msg[:idx]) }As per coding guidelines: "Split assign-and-condition into two statements; prefer separate
derr := action()andif derr != niloverif derr := action(); derr != nil."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/message/message.go` around lines 58 - 60, Split the inline assignment in the if-condition: first call strings.IndexByte(msg, '\n') and assign to idx in its own statement (idx := strings.IndexByte(msg, '\n')), then use a separate if idx >= 0 { msg = strings.TrimSpace(msg[:idx]) } so the behavior of trimming msg up to the first newline is unchanged; update the code around the msg variable and strings.IndexByte usage in message.go accordingly.flow/internal/task/operations/summary.go (1)
63-65: ⚡ Quick winUse separate assignment and error check for JSON unmarshal.
Refactor this inline
ifassignment to two statements to align with the repository’s Go style.Proposed change
if len(w.Info) > 0 { - if err := json.Unmarshal(w.Info, &info); err != nil { - return "", fmt.Errorf("unmarshal firmware info: %w", err) - } + derr := json.Unmarshal(w.Info, &info) + if derr != nil { + return "", fmt.Errorf("unmarshal firmware info: %w", derr) + } }As per coding guidelines: "Split assign-and-condition into two statements; prefer separate
derr := action()andif derr != niloverif derr := action(); derr != nil."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/operations/summary.go` around lines 63 - 65, Refactor the inline if-assignment using json.Unmarshal so the call and error check are two statements: first assign err := json.Unmarshal(w.Info, &info), then follow with if err != nil { return "", fmt.Errorf("unmarshal firmware info: %w", err) }; locate the json.Unmarshal call that currently reads if err := json.Unmarshal(w.Info, &info); err != nil within the summary generation code and replace it with the separate assignment and conditional using the same variables (err and info).flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go (1)
93-96: ⚡ Quick winDocument why
reportis intentionallynilon completion paths.Both terminal calls pass
nilfor report; add a brief inline note/TODO here to make the intentional omission explicit during the transition period.As per coding guidelines, "Document when you have intentionally omitted code that the reader might otherwise expect to be present" and "Add TODO comments for features or nuances not important to implement right away".Proposed tweak
if err := injectExpectationForAll(ctx, typeToTargets, info); err != nil { + // Report is intentionally nil until injectExpectation is migrated to rule-based execution. return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, nil) } + // Report is intentionally nil until injectExpectation is migrated to rule-based execution. return updateFinishedTaskStatus(ctx, reqInfo.TaskID, nil, nil)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go` around lines 93 - 96, The two terminal calls to updateFinishedTaskStatus (updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, nil) and updateFinishedTaskStatus(ctx, reqInfo.TaskID, nil, nil)) intentionally pass nil for the report during this transition; add a concise inline comment/TODO immediately above these return statements explaining that report is omitted intentionally (e.g., "report intentionally nil during migration/feature rollout — populate when X is implemented") and reference the updateFinishedTaskStatus call and the reqInfo.TaskID so reviewers can quickly see the rationale.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@flow/internal/task/store/postgres.go`:
- Around line 199-200: The current code uses an assign-in-if for
taskDao.UpdateTaskReport; split the assignment into two statements: first call
taskDao.UpdateTaskReport(ctx, s.pg.DB, merged) and assign its result to a
variable (e.g., uerr), then check if uerr != nil and return
errors.GRPCErrorInternal(uerr.Error()); this change applies to the block that
currently reads if uerr := taskDao.UpdateTaskReport(ctx, s.pg.DB, merged); uerr
!= nil.
---
Duplicate comments:
In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go`:
- Around line 96-98: The current inline assignment in the activity call should
be split into two statements: first call workflow.ExecuteActivity(actx,
activity.NameUpdateTaskReport, arg).Get(ctx, nil) and assign its error to a
variable (e.g., derr), then use a separate if derr != nil { ... } block to
handle logging; update the code that references taskID and log.Warn().Err(...)
to use that derr variable (preserve the same log message and fields) so it
conforms to the repository rule against assign-and-condition.
- Around line 120-124: When rep.MarshalRaw() fails we currently return only merr
which loses the original operation error stored in err; modify the error
returned in that branch so the original err is preserved as the primary wrapped
error and the marshal failure is added as context. Concretely, in the block that
calls rep.MarshalRaw() (variables err, merr, raw, and call rep.MarshalRaw()),
return a combined error that wraps err with fmt.Errorf and includes the marshal
error message (e.g., wrap err with %w and append the marshal error text) so
callers retain the original terminal error while still seeing marshal failure
details.
- Around line 368-371: The current logic sets componentType from only
stage.Steps[0], which misreports when a stage has mixed component types; update
the computation in the helper where componentType is set to iterate all
step.ComponentType values (from stage.Steps) and determine if they are all
identical—only call devicetypes.ComponentTypeToString when uniform; otherwise
set componentType to a clear sentinel (e.g., "mixed" or empty) to indicate
multiple component types so progress reporting is accurate.
In `@flow/internal/task/store/postgres.go`:
- Around line 167-175: The code is using the global handle s.pg.DB when
reading/updating reports which can bypass a RunInTransaction context; change the
model.GetTask, reportpkg.Merge read path and taskDao.UpdateTaskStatus call to
use the transaction-aware DB handle derived from the incoming ctx instead of
s.pg.DB (e.g., obtain the tx/DB from ctx or the pg wrapper used elsewhere in
this package and pass that handle into model.GetTask and
taskDao.UpdateTaskStatus) so all operations (GetTask, Merge, UpdateTaskStatus)
participate in the caller's transaction boundary.
---
Nitpick comments:
In `@flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go`:
- Around line 93-96: The two terminal calls to updateFinishedTaskStatus
(updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, nil) and
updateFinishedTaskStatus(ctx, reqInfo.TaskID, nil, nil)) intentionally pass nil
for the report during this transition; add a concise inline comment/TODO
immediately above these return statements explaining that report is omitted
intentionally (e.g., "report intentionally nil during migration/feature rollout
— populate when X is implemented") and reference the updateFinishedTaskStatus
call and the reqInfo.TaskID so reviewers can quickly see the rationale.
In `@flow/internal/task/message/message.go`:
- Around line 58-60: Split the inline assignment in the if-condition: first call
strings.IndexByte(msg, '\n') and assign to idx in its own statement (idx :=
strings.IndexByte(msg, '\n')), then use a separate if idx >= 0 { msg =
strings.TrimSpace(msg[:idx]) } so the behavior of trimming msg up to the first
newline is unchanged; update the code around the msg variable and
strings.IndexByte usage in message.go accordingly.
In `@flow/internal/task/operations/summary.go`:
- Around line 63-65: Refactor the inline if-assignment using json.Unmarshal so
the call and error check are two statements: first assign err :=
json.Unmarshal(w.Info, &info), then follow with if err != nil { return "",
fmt.Errorf("unmarshal firmware info: %w", err) }; locate the json.Unmarshal call
that currently reads if err := json.Unmarshal(w.Info, &info); err != nil within
the summary generation code and replace it with the separate assignment and
conditional using the same variables (err and info).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: f873ef9d-89f0-4a50-a3db-7ae9ada31a7a
⛔ Files ignored due to path filters (1)
flow/pkg/proto/v1/flow.pb.gois excluded by!**/*.pb.go,!**/*.pb.go
📒 Files selected for processing (38)
flow/internal/converter/dao/converter.goflow/internal/converter/protobuf/converter.goflow/internal/db/migrations/20260521100000_add_task_report.down.sqlflow/internal/db/migrations/20260521100000_add_task_report.up.sqlflow/internal/db/model/task.goflow/internal/scheduler/taskschedule/dispatcher_test.goflow/internal/scheduler/taskschedule/template.goflow/internal/scheduler/taskschedule/template_test.goflow/internal/task/conflict/store_mock_test.goflow/internal/task/executor/temporalworkflow/activity/activity.goflow/internal/task/executor/temporalworkflow/activity/registry.goflow/internal/task/executor/temporalworkflow/activity/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/bringup.goflow/internal/task/executor/temporalworkflow/workflow/bringup_test.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/helpers.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.goflow/internal/task/manager/manager.goflow/internal/task/manager/notifying_store.goflow/internal/task/message/message.goflow/internal/task/message/message_test.goflow/internal/task/operationrules/rules.goflow/internal/task/operations/summary.goflow/internal/task/operations/summary_test.goflow/internal/task/report/report.goflow/internal/task/report/report_test.goflow/internal/task/store/postgres.goflow/internal/task/store/store.goflow/internal/task/task/task.goflow/proto/v1/flow.proto
✅ Files skipped from review due to trivial changes (2)
- flow/internal/db/migrations/20260521100000_add_task_report.up.sql
- flow/internal/scheduler/taskschedule/template_test.go
jw-nvidia
reviewed
May 27, 2026
| iter := operationrules.NewStageIterator(ruleDef) | ||
| stageIndex := 0 | ||
| for stage := iter.Next(); stage != nil; stage = iter.Next() { | ||
| stageIndex++ |
Contributor
There was a problem hiding this comment.
stageIndex does not seem to be used. If yes, please remove.
jw-nvidia
reviewed
May 27, 2026
| // RunningStage is present only while an operation-rule stage is executing. | ||
| type RunningStage struct { | ||
| Number int `json:"number"` | ||
| ComponentType string `json:"component_type,omitempty"` |
Contributor
There was a problem hiding this comment.
One stage can have multiple component types.
jw-nvidia
reviewed
May 27, 2026
| return opts | ||
| } | ||
|
|
||
| // childWorkflowEntry pairs a launched child workflow future with its component |
Contributor
There was a problem hiding this comment.
Why are the comments for the unexposed structs and methods removed?
jw-nvidia
reviewed
May 27, 2026
| Failed int `json:"failed"` | ||
| } | ||
|
|
||
| type Step struct { |
Contributor
There was a problem hiding this comment.
Can we rename it as "FinishedStage"? Step may cause confusion since we use "Step" in rule for different purpose.
jw-nvidia
reviewed
May 27, 2026
jw-nvidia
left a comment
jw-nvidia
left a comment
Contributor
There was a problem hiding this comment.
Each stage in rule have one or multiple rule steps, each rule step is for one component types. Also same component type may show up in different stages. So there is no point to contain component type information in reported running stage, instead we should consider to include component information for each stage.
kunzhao-nv
force-pushed
the
feat/flow-better-info
branch
2 times, most recently
from
0f9bf00 to
22b20de
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@flow/internal/task/executor/temporalworkflow/workflow/helpers.go`:
- Around line 121-124: The current handler unconditionally overwrites rep.Error
with message.ForFailure(err), which can obliterate the canonical error set
earlier by Tracker.FailStage; update the logic in helpers.go so that when rep !=
nil and err != nil you only assign rep.Error = message.ForFailure(err) if
rep.Error is empty/nil (i.e., preserve an existing canonical error set by
Tracker.FailStage), otherwise leave rep.Error unchanged.
In `@flow/internal/task/report/report.go`:
- Around line 175-178: The MarshalJSON method for Report currently returns (nil,
nil) when the receiver is nil which produces invalid JSON; update
Report.MarshalJSON so that when r == nil it returns []byte("null") and a nil
error to satisfy the "encodes as JSON null" contract and ensure valid JSON
output; locate the MarshalJSON function on type Report and change the
nil-receiver return to return the JSON null bytes instead of nil.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 5202b4ea-535a-4411-861f-0ada9803acdb
⛔ Files ignored due to path filters (1)
flow/pkg/proto/v1/flow.pb.gois excluded by!**/*.pb.go,!**/*.pb.go
📒 Files selected for processing (38)
flow/internal/converter/dao/converter.goflow/internal/converter/protobuf/converter.goflow/internal/db/migrations/20260521100000_add_task_report.down.sqlflow/internal/db/migrations/20260521100000_add_task_report.up.sqlflow/internal/db/model/task.goflow/internal/scheduler/taskschedule/dispatcher_test.goflow/internal/scheduler/taskschedule/template.goflow/internal/scheduler/taskschedule/template_test.goflow/internal/task/conflict/store_mock_test.goflow/internal/task/executor/temporalworkflow/activity/activity.goflow/internal/task/executor/temporalworkflow/activity/registry.goflow/internal/task/executor/temporalworkflow/activity/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/bringup.goflow/internal/task/executor/temporalworkflow/workflow/bringup_test.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/helpers.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.goflow/internal/task/manager/manager.goflow/internal/task/manager/notifying_store.goflow/internal/task/message/message.goflow/internal/task/message/message_test.goflow/internal/task/operationrules/rules.goflow/internal/task/operations/summary.goflow/internal/task/operations/summary_test.goflow/internal/task/report/report.goflow/internal/task/report/report_test.goflow/internal/task/store/postgres.goflow/internal/task/store/store.goflow/internal/task/task/task.goflow/proto/v1/flow.proto
✅ Files skipped from review due to trivial changes (2)
- flow/internal/scheduler/taskschedule/template_test.go
- flow/internal/task/executor/temporalworkflow/workflow/bringup.go
🚧 Files skipped from review as they are similar to previous changes (31)
- flow/internal/task/store/store.go
- flow/internal/task/message/message_test.go
- flow/internal/db/migrations/20260521100000_add_task_report.up.sql
- flow/internal/converter/dao/converter.go
- flow/internal/task/executor/temporalworkflow/activity/activity.go
- flow/internal/task/conflict/store_mock_test.go
- flow/internal/task/executor/temporalworkflow/activity/registry_test.go
- flow/internal/scheduler/taskschedule/dispatcher_test.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go
- flow/internal/task/store/postgres.go
- flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go
- flow/internal/task/operations/summary_test.go
- flow/proto/v1/flow.proto
- flow/internal/db/migrations/20260521100000_add_task_report.down.sql
- flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go
- flow/internal/converter/protobuf/converter.go
- flow/internal/task/manager/manager.go
- flow/internal/task/operationrules/rules.go
- flow/internal/task/message/message.go
- flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go
- flow/internal/scheduler/taskschedule/template.go
- flow/internal/db/model/task.go
- flow/internal/task/task/task.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go
- flow/internal/task/manager/notifying_store.go
- flow/internal/task/operations/summary.go
- flow/internal/task/executor/temporalworkflow/workflow/registry_test.go
- flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go
- flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.go
- flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go
jw-nvidia
reviewed
May 29, 2026
jw-nvidia
approved these changes
May 29, 2026
Tighten task.message to status semantics; description stays user-provided. Add task.report (JSONB) carrying a versioned progress document that mirrors the operation rule: every rule stage maps to a StageRecord and every SequenceStep within it maps to a StepRecord at the same index. Steps whose ComponentType has no targets in the task start as StatusSkipped; the rest advance pending -> running -> completed|failed via report.Tracker at stage boundaries. Under the fail-fast activity contract a stage failure marks every still-running sibling step failed because their results are unobserved and reporting them as completed would be a guess. StepRecord carries TotalComponents, filled at NewInitial from task.attributes.components_by_type. The API task representation does not surface that map, so the report has to be self-describing or a client cannot size each step's work. CompletedComponents and FailedComponents are reserved on the wire (omitempty) for a future best-effort activity contract that reports per-component outcomes; they are not written under the current fail-fast contract and so do not appear in the JSON payload today. Pending and processed counts are derivable on the client (Pending = TotalComponents - Completed - Failed; processed = Completed + Failed) and are not stored. Workflows emit full Tracker snapshots on every stage transition, so the store treats writes as authoritative replacements: a non-empty report payload overwrites task.report, an empty payload leaves it untouched. No read-modify-write is performed in either UpdateTaskStatus or UpdateTaskReport, which closes the lost-update race that a merge helper would have introduced. Tracker.FailStage is the canonical writer for the report's task-level error (first-failure-wins); updateFinishedTaskStatus only fills it in when the Tracker did not, so a workflow that constructs a report outside the rule-based path still records its terminal error. The status updater and the report updater are wired through the executor stack as separate parameters rather than via a runtime type assertion. Each role is therefore explicit at every call site and implementations are free to differ; today both are served by the notifying-store wrapper, which declares both interface assertions statically. Signed-off-by: Kun Zhao <kunzhao@nvidia.com>
kunzhao-nv
force-pushed
the
feat/flow-better-info
branch
from
22b20de to
4313cf9
Compare
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
flow/internal/task/report/report.go (1)
251-305: ⚡ Quick winGuard
CompleteStage/FailStageagainst non-running and terminal stages.
BeginStagecorrectly refuses to act unless the stage isStatusPending, butCompleteStageandFailStagemutate unconditionally. Two consequences follow from the documented idempotency contract:
CompleteStageon a still-StatusPendingstage marks the stageStatusCompletedwhile leaving its steps inStatusPending(none areStatusRunningto finalize), yielding an inconsistent snapshot.- Either method can overwrite a stage that is already in a terminal state, so a late/duplicated transition could clobber a prior
StatusFailedoutcome.Constraining each transition to its expected predecessor state keeps snapshots consistent and the transitions genuinely idempotent.
♻️ Proposed guards
func (t *Tracker) CompleteStage(stageNum int, now time.Time) { if t == nil || t.Report == nil { return } stage := t.findStage(stageNum) - if stage == nil { + if stage == nil || stage.Status != StatusRunning { return }func (t *Tracker) FailStage(stageNum int, stageErr error, now time.Time) { if t == nil || t.Report == nil { return } stage := t.findStage(stageNum) - if stage == nil { + if stage == nil || stage.Status == StatusCompleted || stage.Status == StatusFailed { return }Please confirm the workflow never legitimately fails a stage still in
StatusPending(e.g. a pre-operation failure beforeBeginStage); if it does, relax theFailStageguard to exclude only terminal states.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flow/internal/task/report/report.go` around lines 251 - 305, CompleteStage and FailStage currently mutate stages unconditionally, causing inconsistent snapshots and allowing terminal states to be clobbered; update these methods to check the current stage.Status before mutating: in CompleteStage (function CompleteStage) return early unless stage.Status == StatusRunning so you only mark running stages completed and finalize only running steps; in FailStage (function FailStage) return early if stage.Status is already a terminal state (StatusCompleted or StatusFailed) so you don’t overwrite prior outcomes—if your workflow can legitimately fail a StatusPending stage, relax this guard to allow non-terminal statuses (i.e., allow Pending or Running) but still refuse to touch terminal ones; ensure stage.FinishedAt, stage.Status, stage.Error and step Status/FinishedAt/Error are only set when the guard permits the transition.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@flow/internal/task/report/report.go`:
- Around line 251-305: CompleteStage and FailStage currently mutate stages
unconditionally, causing inconsistent snapshots and allowing terminal states to
be clobbered; update these methods to check the current stage.Status before
mutating: in CompleteStage (function CompleteStage) return early unless
stage.Status == StatusRunning so you only mark running stages completed and
finalize only running steps; in FailStage (function FailStage) return early if
stage.Status is already a terminal state (StatusCompleted or StatusFailed) so
you don’t overwrite prior outcomes—if your workflow can legitimately fail a
StatusPending stage, relax this guard to allow non-terminal statuses (i.e.,
allow Pending or Running) but still refuse to touch terminal ones; ensure
stage.FinishedAt, stage.Status, stage.Error and step Status/FinishedAt/Error are
only set when the guard permits the transition.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 8c46f3ac-293e-48d3-88b1-ca0d81672120
⛔ Files ignored due to path filters (1)
flow/pkg/proto/v1/flow.pb.gois excluded by!**/*.pb.go,!**/*.pb.go
📒 Files selected for processing (42)
flow/internal/converter/dao/converter.goflow/internal/converter/protobuf/converter.goflow/internal/db/migrations/20260521100000_add_task_report.down.sqlflow/internal/db/migrations/20260521100000_add_task_report.up.sqlflow/internal/db/model/task.goflow/internal/scheduler/taskschedule/dispatcher_test.goflow/internal/scheduler/taskschedule/template.goflow/internal/scheduler/taskschedule/template_test.goflow/internal/task/conflict/store_mock_test.goflow/internal/task/executor/executor.goflow/internal/task/executor/temporalworkflow/README.mdflow/internal/task/executor/temporalworkflow/activity/activity.goflow/internal/task/executor/temporalworkflow/activity/activity_test.goflow/internal/task/executor/temporalworkflow/activity/registry.goflow/internal/task/executor/temporalworkflow/activity/registry_test.goflow/internal/task/executor/temporalworkflow/manager/manager.goflow/internal/task/executor/temporalworkflow/workflow/bringup.goflow/internal/task/executor/temporalworkflow/workflow/bringup_test.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.goflow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/helpers.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation.goflow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.goflow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.goflow/internal/task/executor/temporalworkflow/workflow/registry_test.goflow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.goflow/internal/task/manager/manager.goflow/internal/task/manager/notifying_store.goflow/internal/task/message/message.goflow/internal/task/message/message_test.goflow/internal/task/operationrules/rules.goflow/internal/task/operations/summary.goflow/internal/task/operations/summary_test.goflow/internal/task/report/report.goflow/internal/task/report/report_test.goflow/internal/task/store/postgres.goflow/internal/task/store/store.goflow/internal/task/task/task.goflow/proto/v1/flow.proto
✅ Files skipped from review due to trivial changes (2)
- flow/internal/task/executor/temporalworkflow/README.md
- flow/internal/scheduler/taskschedule/template_test.go
🚧 Files skipped from review as they are similar to previous changes (32)
- flow/internal/db/migrations/20260521100000_add_task_report.down.sql
- flow/internal/converter/dao/converter.go
- flow/internal/task/operationrules/rules.go
- flow/internal/scheduler/taskschedule/dispatcher_test.go
- flow/internal/task/message/message_test.go
- flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go
- flow/internal/task/executor/temporalworkflow/workflow/bringup.go
- flow/internal/converter/protobuf/converter.go
- flow/internal/task/store/store.go
- flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go
- flow/internal/task/executor/temporalworkflow/workflow/registry_test.go
- flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go
- flow/internal/task/executor/temporalworkflow/activity/activity.go
- flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go
- flow/internal/task/operations/summary_test.go
- flow/internal/scheduler/taskschedule/template.go
- flow/internal/db/model/task.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go
- flow/internal/task/manager/notifying_store.go
- flow/internal/task/operations/summary.go
- flow/internal/task/message/message.go
- flow/internal/task/task/task.go
- flow/internal/task/executor/temporalworkflow/activity/registry_test.go
- flow/proto/v1/flow.proto
- flow/internal/task/report/report_test.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go
- flow/internal/db/migrations/20260521100000_add_task_report.up.sql
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go
- flow/internal/task/executor/temporalworkflow/workflow/helpers.go
- flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go
- flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go
- flow/internal/task/store/postgres.go
Signed-off-by: Kun Zhao <kunzhao@nvidia.com> # Conflicts: # flow/internal/task/executor/temporalworkflow/activity/registry_test.go
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Update
Type of Change
Services Affected
Related Issues (Optional)
Breaking Changes
Testing
Additional Notes