feat: Improve Flow task description, message, and progress report

flow/internal/converter/dao/converter.go

@@ -177,6 +177,7 @@ func TaskFrom(dao *model.Task) *taskdef.Task {
 		ExecutionID:    dao.ExecutionID,
 		Status:         dao.Status,
 		Message:        dao.Message,
+		Report:         dao.Report,
 		AppliedRuleID:  dao.AppliedRuleID,
 		CreatedAt:      dao.CreatedAt,
 		UpdatedAt:      dao.UpdatedAt,
@@ -312,6 +313,7 @@ func TaskTo(task *taskdef.Task) *model.Task {
 		ExecutionID:    task.ExecutionID,
 		Status:         task.Status,
 		Message:        task.Message,
+		Report:         task.Report,
 		AppliedRuleID:  task.AppliedRuleID,
 		QueueExpiresAt: task.QueueExpiresAt,
 	}

flow/internal/converter/protobuf/converter.go

@@ -466,6 +466,7 @@ func TaskTo(task *taskdef.Task) *pb.Task {
 		ExecutionId:    task.ExecutionID,
 		Status:         TaskStatusTo(task.Status),
 		Message:        task.Message,
+		Report:         string(task.Report),
 		CreatedAt:      timestamppb.New(task.CreatedAt),
 		UpdatedAt:      timestamppb.New(task.UpdatedAt),
 	}

flow/internal/db/migrations/20260521100000_add_task_report.down.sql

@@ -0,0 +1,5 @@
+-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+-- SPDX-License-Identifier: Apache-2.0
+
+ALTER TABLE public.task
+    DROP COLUMN IF EXISTS report;

flow/internal/db/migrations/20260521100000_add_task_report.up.sql

@@ -0,0 +1,5 @@
+-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+-- SPDX-License-Identifier: Apache-2.0
+
+ALTER TABLE public.task
+    ADD COLUMN report jsonb;

flow/internal/db/model/task.go

@@ -36,6 +36,7 @@ type Task struct {
 	ExecutionID   string                    `bun:"execution_id,notnull"`
 	Status        taskcommon.TaskStatus     `bun:"status,type:varchar(32),notnull"`
 	Message       string                    `bun:"message,nullzero"`
+	Report        json.RawMessage           `bun:"report,type:jsonb"`
 	AppliedRuleID *uuid.UUID                `bun:"applied_rule_id,type:uuid"` // Which operation rule was applied
 	CreatedAt     time.Time                 `bun:"created_at,nullzero,notnull,default:current_timestamp"`
 	UpdatedAt     time.Time                 `bun:"updated_at,nullzero,notnull,default:current_timestamp"`
@@ -78,17 +79,23 @@ func (t *Task) UpdateScheduledTask(
 }
 
 // UpdateTaskStatus updates the status of the task.
+// report, when non-nil, replaces the stored report column.
 func (t *Task) UpdateTaskStatus(
 	ctx context.Context,
 	idb bun.IDB,
 	status taskcommon.TaskStatus,
 	message string,
+	report json.RawMessage,
 ) error {
 	t.Status = status
 	t.Message = message
 	t.UpdatedAt = time.Now().UTC()
 
 	columns := []string{"status", "message", "updated_at", "finished_at"}
+	if report != nil {
+		t.Report = report
+		columns = append(columns, "report")
+	}
 
 	if status == taskcommon.TaskStatusRunning && t.StartedAt == nil {
 		t.StartedAt = &t.UpdatedAt
@@ -109,6 +116,24 @@ func (t *Task) UpdateTaskStatus(
 	return err
 }
 
+// UpdateTaskReport updates the structured report without changing status or message.
+func (t *Task) UpdateTaskReport(
+	ctx context.Context,
+	idb bun.IDB,
+	report json.RawMessage,
+) error {
+	t.Report = report
+	t.UpdatedAt = time.Now().UTC()
+
+	_, err := idb.NewUpdate().
+		Model(t).
+		Column("report", "updated_at").
+		Where("id = ?", t.ID).
+		Exec(ctx)
+
+	return err
+}
+
 // taskComponentIDFilter restricts a query to tasks whose attributes
 // reference a specific component UUID. The UUID may live under any bucket
 // of attributes.components_by_type, which rules out a fixed @> check

flow/internal/scheduler/taskschedule/dispatcher_test.go

@@ -175,6 +175,10 @@ func (m *mockTaskStore) UpdateTaskStatus(_ context.Context, _ *taskdef.TaskStatu
 	panic("mockTaskStore.UpdateTaskStatus: not implemented")
 }
 
+func (m *mockTaskStore) UpdateTaskReport(_ context.Context, _ *taskdef.TaskReportUpdate) error {
+	panic("mockTaskStore.UpdateTaskReport: not implemented")
+}
+
 func (m *mockTaskStore) ListActiveTasksForRack(_ context.Context, _ uuid.UUID) ([]*taskdef.Task, error) {
 	panic("mockTaskStore.ListActiveTasksForRack: not implemented")
 }

flow/internal/scheduler/taskschedule/template.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/NVIDIA/infra-controller-rest/flow/internal/operation"
 	taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common"
+	"github.com/NVIDIA/infra-controller-rest/flow/internal/task/operations"
 )
 
 // TaskTemplate is the JSON-serialized operation stored in task_schedule.operation_template.
@@ -97,61 +98,15 @@ func OptionsFromTemplate(raw json.RawMessage) (TemplateOptions, error) {
 // (e.g. "POWER_ON", "BRING_UP"). description is a short English phrase for
 // display (e.g. "Power Off (forced)", "Upgrade Firmware to v2.1.0").
 func SummaryFromTemplate(templateJSON json.RawMessage) (opType, description string, err error) {
-	var tmpl TaskTemplate
-	if err = json.Unmarshal(templateJSON, &tmpl); err != nil {
-		return "", "", fmt.Errorf("unmarshal operation_template: %w", err)
+	w, err := WrapperFromTemplate(templateJSON)
+	if err != nil {
+		return "", "", err
 	}
 
-	switch tmpl.Type {
-	case taskcommon.TaskTypePowerControl:
-		switch tmpl.Code {
-		case taskcommon.OpCodePowerControlPowerOn, taskcommon.OpCodePowerControlForcePowerOn:
-			return "POWER_ON", "Power On", nil
-		case taskcommon.OpCodePowerControlPowerOff:
-			return "POWER_OFF", "Power Off", nil
-		case taskcommon.OpCodePowerControlForcePowerOff:
-			return "POWER_OFF", "Power Off (forced)", nil
-		case taskcommon.OpCodePowerControlRestart, taskcommon.OpCodePowerControlWarmReset:
-			return "POWER_RESET", "Power Reset", nil
-		case taskcommon.OpCodePowerControlForceRestart, taskcommon.OpCodePowerControlColdReset:
-			return "POWER_RESET", "Power Reset (forced)", nil
-		default:
-			return "POWER_CONTROL", tmpl.Code, nil
-		}
-
-	case taskcommon.TaskTypeBringUp:
-		if tmpl.Code == taskcommon.OpCodeIngest {
-			return "INGEST", "Ingest", nil
-		}
-		return "BRING_UP", "Bring Up", nil
-
-	case taskcommon.TaskTypeFirmwareControl:
-		var info struct {
-			TargetVersion string `json:"target_version"`
-		}
-		if len(tmpl.Info) > 0 {
-			if err = json.Unmarshal(tmpl.Info, &info); err != nil {
-				return "", "", fmt.Errorf("unmarshal firmware info: %w", err)
-			}
-		}
-		switch tmpl.Code {
-		case taskcommon.OpCodeFirmwareControlUpgrade:
-			if info.TargetVersion != "" {
-				return "UPGRADE_FIRMWARE", "Upgrade Firmware to " + info.TargetVersion, nil
-			}
-			return "UPGRADE_FIRMWARE", "Upgrade Firmware", nil
-		case taskcommon.OpCodeFirmwareControlDowngrade:
-			if info.TargetVersion != "" {
-				return "DOWNGRADE_FIRMWARE", "Downgrade Firmware to " + info.TargetVersion, nil
-			}
-			return "DOWNGRADE_FIRMWARE", "Downgrade Firmware", nil
-		case taskcommon.OpCodeFirmwareControlRollback:
-			return "ROLLBACK_FIRMWARE", "Rollback Firmware", nil
-		default: // unrecognized
-			return "FIRMWARE_CONTROL", tmpl.Code, nil
-		}
-
-	default:
-		return string(tmpl.Type), tmpl.Code, nil
+	description, err = operations.SummaryFromWrapper(w)
+	if err != nil {
+		return "", "", err
 	}
+
+	return operations.OperationTypeFromWrapper(w), description, nil
 }

flow/internal/scheduler/taskschedule/template_test.go

@@ -214,7 +214,7 @@ func TestSummaryFromTemplate(t *testing.T) {
 		"unknown power control code": {
 			input:      mustMarshalTemplate(taskcommon.TaskTypePowerControl, "bmc_cycle", emptyInfo),
 			wantOpType: "POWER_CONTROL",
-			wantDesc:   "bmc_cycle",
+			wantDesc:   "bmc cycle",
 		},
 
 		// ── bring-up / ingest ────────────────────────────────────────────────
@@ -267,8 +267,8 @@ func TestSummaryFromTemplate(t *testing.T) {
 		// ── unknown type ──────────────────────────────────────────────────────
 		"unknown operation type": {
 			input:      mustMarshalTemplate(taskcommon.TaskTypeInjectExpectation, "inject", emptyInfo),
-			wantOpType: string(taskcommon.TaskTypeInjectExpectation),
-			wantDesc:   "inject",
+			wantOpType: "INJECT_EXPECTATION",
+			wantDesc:   "Inject Expectation",
 		},
 	}
 

flow/internal/task/conflict/store_mock_test.go

@@ -65,6 +65,10 @@ func (m *mockStore) ListRacksWithWaitingTasks(ctx context.Context) ([]uuid.UUID,
 	return m.racksWithWaiting, nil
 }
 
+func (m *mockStore) UpdateTaskReport(_ context.Context, _ *taskdef.TaskReportUpdate) error {
+	return nil
+}
+
 func (m *mockStore) UpdateTaskStatus(ctx context.Context, arg *taskdef.TaskStatusUpdate) error {
 	if m.updateStatusErr != nil {
 		return m.updateStatusErr

flow/internal/task/executor/executor.go

@@ -30,16 +30,25 @@ type Executor interface {
 type ExecutorConfig interface {
 	Validate() error
 	// Build constructs the Executor. updater receives task status transitions
-	// from the execution engine back to the store. It must not be nil.
-	Build(ctx context.Context, updater task.TaskStatusUpdater) (Executor, error)
+	// from the execution engine back to the store and must not be nil.
+	// reportUpdater receives in-flight report snapshots and may be nil; when
+	// nil, activities that need it return an error at invocation time.
+	Build(
+		ctx context.Context,
+		updater task.TaskStatusUpdater,
+		reportUpdater task.TaskReportUpdater,
+	) (Executor, error)
 }
 
 // New validates the config and builds the Executor, wiring updater so the
 // engine can report task status changes without importing store packages.
+// reportUpdater is plumbed through the same call to persist progress
+// snapshots; nil is allowed and disables in-flight reporting.
 func New(
 	ctx context.Context,
 	executorConfig ExecutorConfig,
 	updater task.TaskStatusUpdater,
+	reportUpdater task.TaskReportUpdater,
 ) (Executor, error) {
 	if executorConfig == nil {
 		return nil, fmt.Errorf("executor config is required")
@@ -53,5 +62,5 @@ func New(
 		return nil, fmt.Errorf("task status updater is required")
 	}
 
-	return executorConfig.Build(ctx, updater)
+	return executorConfig.Build(ctx, updater, reportUpdater)
 }

flow/internal/task/executor/temporalworkflow/README.md

@@ -50,7 +50,7 @@ ExecutionRequest{OperationType, OperationInfo}
 
 **Workflow registry** (`workflow/registry.go`): uses `init()` self-registration. Task-dispatched workflow files call `registerTaskWorkflow[T, *T](taskType, name, fn)`, which derives the timeout and builds the `Unmarshal` closure automatically. Internal workflows (those without a `TaskType`) call `register(WorkflowDescriptor{...})` directly. Nothing needs to be added to a central list — the registry is populated automatically at startup.
 
-**Activity registry** (`activity/registry.go`): uses per-instance dependency injection. `Build()` creates an `*Activities` value via `activity.New(updater, registry)` and calls `acts.All()` to obtain the name → bound-method map, then registers each entry with the Temporal worker via `RegisterActivityWithOptions(fn, {Name: name})`. Because activities are methods on `*Activities`, each manager instance holds its own isolated copy of the dependencies — no shared mutable globals.
+**Activity registry** (`activity/registry.go`): uses per-instance dependency injection. `Build()` creates an `*Activities` value via `activity.New(updater, reportUpdater, registry)` and calls `acts.All()` to obtain the name → bound-method map, then registers each entry with the Temporal worker via `RegisterActivityWithOptions(fn, {Name: name})`. Status and report updaters are wired as separate parameters so each role is explicit at the call site. Because activities are methods on `*Activities`, each manager instance holds its own isolated copy of the dependencies — no shared mutable globals.
 
 ## Adding a New Operation
 

flow/internal/task/executor/temporalworkflow/activity/activity.go

@@ -22,6 +22,7 @@ const (
 	NamePowerControl              = "PowerControl"
 	NameGetPowerStatus            = "GetPowerStatus"
 	NameUpdateTaskStatus          = "UpdateTaskStatus"
+	NameUpdateTaskReport          = "UpdateTaskReport"
 	NameFirmwareControl           = "FirmwareControl"
 	NameGetFirmwareStatus         = "GetFirmwareStatus"
 	NameBringUpControl            = "BringUpControl"
@@ -89,6 +90,23 @@ func (a *Activities) UpdateTaskStatus(
 	return a.updater.UpdateTaskStatus(ctx, arg)
 }
 
+// UpdateTaskReport is a Temporal activity that merges a structured report
+// snapshot without changing status or message.
+func (a *Activities) UpdateTaskReport(
+	ctx context.Context,
+	arg *task.TaskReportUpdate,
+) error {
+	if a.reportUpdater == nil {
+		return fmt.Errorf("task report updater is not configured")
+	}
+
+	if arg == nil || arg.ID == uuid.Nil {
+		return fmt.Errorf("invalid task identifier")
+	}
+
+	return a.reportUpdater.UpdateTaskReport(ctx, arg)
+}
+
 // FirmwareControl initiates firmware update without waiting for completion.
 // This activity returns immediately after the update request is accepted.
 func (a *Activities) FirmwareControl(

flow/internal/task/executor/temporalworkflow/activity/activity_test.go

@@ -21,7 +21,7 @@ import (
 )
 
 func TestActivitiesReturnErrorWhenComponentManagerRegistryIsMissing(t *testing.T) {
-	acts := New(nil, nil)
+	acts := New(nil, nil, nil)
 
 	for name, call := range activityCallsForMissingManagerTest(t, acts) {
 		t.Run(name, func(t *testing.T) {
@@ -40,7 +40,7 @@ func TestActivitiesReturnErrorWhenComponentManagerIsMissing(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	acts := New(nil, registry)
+	acts := New(nil, nil, registry)
 
 	for name, call := range activityCallsForMissingManagerTest(t, acts) {
 		t.Run(name, func(t *testing.T) {
@@ -299,7 +299,7 @@ func newCapabilityTestActivities(
 	)
 	require.NoError(t, err)
 
-	return New(nil, registry), manager
+	return New(nil, nil, registry), manager
 }
 
 type descriptorOnlyManager struct {
@@ -341,7 +341,7 @@ func newDescriptorOnlyActivities(
 	)
 	require.NoError(t, err)
 
-	return New(nil, registry)
+	return New(nil, nil, registry)
 }
 
 func newActivityTestTarget() common.Target {

flow/internal/task/executor/temporalworkflow/activity/registry.go

@@ -14,20 +14,22 @@ import (
 // independent, multiple managers can coexist in the same process without
 // sharing mutable state.
 type Activities struct {
-	updater  task.TaskStatusUpdater
-	registry *componentmanager.Registry
+	updater       task.TaskStatusUpdater
+	reportUpdater task.TaskReportUpdater
+	registry      *componentmanager.Registry
 }
 
-// New creates an Activities instance bound to the given status updater and
-// component manager registry. Either argument may be nil; activity calls that
-// require the missing dependency will return an error at invocation time.
+// New creates an Activities instance. Any argument may be nil; activity
+// calls that need a missing dependency return an error at invocation time.
 func New(
 	updater task.TaskStatusUpdater,
+	reportUpdater task.TaskReportUpdater,
 	registry *componentmanager.Registry,
 ) *Activities {
 	return &Activities{
-		updater:  updater,
-		registry: registry,
+		updater:       updater,
+		reportUpdater: reportUpdater,
+		registry:      registry,
 	}
 }
 
@@ -41,6 +43,7 @@ func (a *Activities) All() map[string]any {
 		NamePowerControl:              a.PowerControl,
 		NameGetPowerStatus:            a.GetPowerStatus,
 		NameUpdateTaskStatus:          a.UpdateTaskStatus,
+		NameUpdateTaskReport:          a.UpdateTaskReport,
 		NameFirmwareControl:           a.FirmwareControl,
 		NameGetFirmwareStatus:         a.GetFirmwareStatus,
 		NameBringUpControl:            a.BringUpControl,