feat(task): implement GraphResolver with fan-in cycle pruning and migrate monorepo to Task System v3 (#976)

* feat(task): implement 4-phase GraphResolver, concrete edge tracking, and memory lifecycle in LocalRunner

* refactor(inspection): modernize inspection taskbase and runner for edge attributes

* refactor(task): migrate inspection tasks across monorepo to new task system

* feat(task): implement priority-based fan-in pruning and cycle detection in GraphResolver

* fix(review): address review comments on formtask aliases, inventory task multi-stage, and cycle tests

* feat(task): support stage-aware fan-in binding in TaskGraphMetadata and TaskSet

* feat(task): split cyclic multi-stage consumer tasks into DAG stages in GraphResolver

* test(inspection): verify end-to-end multi-stage inventory aggregation with cyclic producer

* fix(task): resolve LocalRunner stage tracking data race and improve naming

* refactor(task): eliminate dead BoundReferenceIDsWithTag and normalize task-scoped tag metadata

* refactor(task): decompose cycle resolution helpers and add edge routing tests

* refactor(task): clean up taskset doc comment, runner identifiers, and taskid parameter naming

* refactor(task): decompose cycle resolution helpers and extract graph utilities into graphresolver_graphutil.go

* test(task): add upstream PtP edge routing test for split multi-stage consumer

* refactor(task): eliminate obsolete refToTask and refToImplID maps in cycle resolution

* refactor(task): decompose cycle resolution and stage expansion functions

* refactor(task): standardize PointToPoint identifiers and fix inverted test description

* refactor(task): rename TaskEdge endpoints to SourceImplID and TargetImplID

* refactor(task): extract stage expansion and rerouting into graphresolver_stage.go

* fix(task): reroute fan-in edges from split producer and ensure deterministic stage edges

* refactor(task): align stage identifier names and test mock field naming

* refactor(task): extract stage tests into graphresolver_stage_test.go and co-locate stageTask

* fix(task): prevent premature deletion of self-loop stage results and add runner test

* refactor(task): decompose rerouteFanInEdges in graphresolver_stage.go

* refactor(task): standardize remainingStagesByRefID and boundFanInRefIDsByTaskImpl identifiers

* test(task): add unit tests for graphresolver_graphutil.go

* fix(task): replace scalar cmp.Diff with direct equality in runner_test.go and add stage assertions

* refactor(task): eliminate abbreviations and align identifiers in stage resolution and tests

* refactor(task): rename rem and remDep to remainingDependents in LocalRunner

* refactor(task): align stage edge identifiers and non-split naming in stage resolution

* refactor(task): flatten control flow nesting in LocalRunner cleanup

* refactor(task): fix point-to-point doc comment and disambiguate split consumer IDs

* refactor(task): rename BoundReferenceIDsForTaskWithTag to BoundReferenceIDsForTaskImplWithTag

* refactor(task): align stageTaskPairs, producer identifiers, and table-driven graphutil tests

* refactor(task): prune cyclic fan-in candidate edges into single-stage DAG and unify dependency scopes

Author: kyasbal

docs/en/khi-task-system-concept.md

@@ -112,7 +112,7 @@ In KHI, connections (edges) between tasks in the DAG are represented by the `Dep
   Represents a direct 1-to-1 dependency on a specific task reference (`taskID.Ref()`). Downstream tasks read the upstream task's return value using `coretask.GetTaskResult(ctx, ref)`.
 - **Tag Fan-In (`TagReference[T]`)**:
   Represents a 1-to-N aggregated dependency. Producer tasks declare the tags they provide using the `coretask.ProvidesTag(tag, opts...)` label option. You can optionally specify `coretask.WithTagPriority(priority)` to assign precedence to the producer's contribution (default: 100, where lower numerical values indicate higher precedence). A consumer task declares a dependency on the tag using `tag.Ref()`. During execution, the consumer retrieves a combined slice of results (`[]T`) from all active producer tasks using `coretask.GetTaskResultsWithTag(ctx, tag.Ref())`. This allows new log parsers or metadata producers to be added without modifying downstream consumer tasks.
-  When cross-inventory dependencies between multiple producers cause circular dependencies, pure aggregator tasks annotated with `coretask.AllowMultiStageExecution()` can be split into multiple execution stages by the graph resolver to automatically resolve cycles. For details on prerequisites and resolution mechanisms, see [6. Prerequisites of Fan-In Cycles and Graph Stabilization via Priority](#6-prerequisites-of-fan-in-cycles-and-graph-stabilization-via-priority).
+  When cross-inventory dependencies between multiple producers cause circular dependencies, the graph resolver deterministically prunes candidate fan-in edges that form cycles, automatically resolving the circular dependency into a single-stage DAG. For details on prerequisites and resolution mechanisms, see [6. Prerequisites of Fan-In Cycles and Graph Stabilization via Priority](#6-prerequisites-of-fan-in-cycles-and-graph-stabilization-via-priority).
 
 #### 2. Edge Kind: Data vs Order-Only
 
@@ -200,11 +200,7 @@ KHI achieves an always unique, deterministic, and stable graph through the follo
    Producer tasks declare their contribution certainty and priority using `ProvidesTag(tag, WithTagPriority(priority))` (default: 100, where lower numerical values indicate higher precedence).
    - For example: A parser providing definitive metadata early in execution has high precedence (`Priority: 10`), whereas a parser supplementing metadata later as a byproduct of parsing has low precedence (`Priority: 100`).
 2. **Priority-Based Deterministic Pruning**:
-   When a cycle is detected across fan-in dependencies, the graph resolver deterministically prunes the fan-in edge with the **lowest priority (highest numerical value)** within the cycle, restoring an acyclic DAG.
-3. **Strict Fail-Fast on Priority Ties**:
-   If edges in a cycle share identical priorities and the resolver cannot deterministically pick which edge to prune, it does not guess. Graph resolution fails fast immediately with an error.
-4. **Data Preservation via Multi-Stage Execution (`AllowMultiStageExecution`)**:
-   For pure, side-effect-free aggregator tasks (such as in-memory inventory aggregators), annotating them with `AllowMultiStageExecution()` permits the resolver to automatically split and clone them into early and late execution stages. The early stage receives high-priority inputs, while the late stage collects feedback inputs from dependent parsers, safely resolving cycles without dropping data.
+   When candidate fan-in edges are evaluated, the graph resolver considers producers in order of priority and deterministically prunes any candidate edge that would form a cycle, yielding a safe, acyclic DAG.
 
 ---
 

docs/en/task-system/01-syntax-and-modes.md

@@ -89,15 +89,13 @@ var ParserTaskA = task.NewTask(
 )
 
 // 3. Consumer task aggregates all active producers with GetTaskResultsWithTag
-// Pure aggregator tasks can specify AllowMultiStageExecution() to permit multi-stage execution
 var AggregatorTask = task.NewTask(
     AggregatorTaskID,
     []coretask.Dependency{LogItemTag.Ref()},
     func(ctx context.Context) ([]*LogItem, error) {
         items := coretask.GetTaskResultsWithTag(ctx, LogItemTag.Ref())
         return items, nil
     },
-    coretask.AllowMultiStageExecution(),
 )
 ```
 
@@ -110,9 +108,7 @@ When using fan-in aggregation, circular dependencies (cycles) can arise under th
 2. **Deterministic Pruning via Priority for a Stable Graph**:
    Arbitrarily cutting edges to break cycles causes execution order and data flow to fluctuate based on task registration order, producing an unreproducible, unstable graph.
    - `coretask.WithTagPriority(priority)` (default: `DefaultTagPriority = 100`, where lower numbers indicate higher precedence) lets producers declare the certainty and priority of their contribution.
-   - The graph resolver deterministically prunes the lowest-priority fan-in edge within the cycle, producing an always unique and stable graph. If priorities tie within a cycle and the choice is ambiguous, the resolver does not guess and fails fast with an error.
-3. **Multi-Stage Execution (`coretask.AllowMultiStageExecution`)**:
-   To avoid losing data when pruning feedback edges, pure side-effect-free aggregator tasks should declare `AllowMultiStageExecution()`. The resolver automatically splits and clones the aggregator into an early stage (passing high-priority inputs to parsers) and a late stage (collecting feedback outputs after parsers complete). If an unlabelled task requires multi-stage execution to resolve a cycle, graph resolution fails fast with an error.
+   - The graph resolver deterministically prunes candidate fan-in edges that form cycles, consistently producing a safe, unique, and stable single-stage DAG.
 
 For architectural details, see [Concept Guide: 6. Prerequisites of Fan-In Cycles and Graph Stabilization via Priority](../khi-task-system-concept.md#6-prerequisites-of-fan-in-cycles-and-graph-stabilization-via-priority).
 

docs/ja/khi-task-system-concept.md

@@ -112,7 +112,7 @@ KHI では、DAG 内のタスク間の接続（エッジ）は `Dependency` イ
   特定のタスク参照に対する直接的な 1 対 1 の依存関係 (`taskID.Ref()`) を表します。後続のタスクは `coretask.GetTaskResult(ctx, ref)` を使用して先行タスクの型付きの戻り値を取得します。
 - **タグによるファンイン (`TagReference[T]`)**:
   1 対 N の集約的な依存関係を表します。プロデューサタスクは `coretask.ProvidesTag(tag, opts...)` ラベルオプションを使用して提供するタグを宣言します。オプションとして `coretask.WithTagPriority(priority)` を指定することで、プロデューサごとの寄与度や優先度を付与できます。デフォルト値は 100 であり、数値が小さいほど高優先度として扱われます。コンシューマタスクは `tag.Ref()` を指定してタグに依存します。実行時、コンシューマは `coretask.GetTaskResultsWithTag(ctx, tag.Ref())` を呼び出すことで、バインドされたすべてのプロデューサタスクの結果のスライス (`[]T`) をまとめて取得します。これにより、ダウンストリームのタスクを変更することなく、新しいログパーサーやメタデータプロデューサを追加できます。
-  また、クロスインベントリ依存など複数のプロデューサ間で相互依存が生じる場合、純粋な集約タスクに `coretask.AllowMultiStageExecution()` を付与することで、グラフ解決時にタスクをマルチステージに分割して循環依存を自動解消できます。詳細な前提条件と解決メカニズムは「[6. ファンインにおける循環依存の前提条件と Priority によるグラフ安定化](#6-ファンインにおける循環依存の前提条件と-priority-によるグラフ安定化)」を参照してください。
+  また、クロスインベントリ依存など複数のプロデューサ間で相互依存が生じる場合、グラフリゾルバはサイクルを構成するファンインエッジを決定論的に除外することで、単一ステージで循環依存を自動解消します。詳細な前提条件と解決メカニズムは「[6. ファンインにおける循環依存の前提条件と Priority によるグラフ安定化](#6-ファンインにおける循環依存の前提条件と-priority-によるグラフ安定化)」を参照してください。
 
 #### 2. エッジ種別: データエッジ vs 順序制御エッジ (Order-Only)
 
@@ -200,11 +200,7 @@ KHI では以下の仕組みによって、常に一意かつ決定論的で安
    プロデューサタスクは `ProvidesTag(tag, WithTagPriority(priority))` を使用して、そのタグに対して自身が提供するデータの信頼度や決定度を宣言します。デフォルトは 100 であり、数値が小さいほど高優先度として扱われます。
    たとえば起動時に確定的なメタデータを提供するパーサーには `Priority: 10` など高い優先度を付与し、ログ解析の副産物として事後的にメタデータを補完するパーサーには `Priority: 100` などの低い優先度を付与します。
 2. **優先度に基づく決定論的枝刈り**:
-   ファンイン依存関係によって循環が検出された場合、グラフリゾルバはサイクルを構成するファンインエッジの中から最も優先度の低い、すなわち数値の大きいエッジを決定論的に枝刈りし、非循環 DAG を再構築します。
-3. **同着時の Fail-Fast 保証**:
-   サイクル内に同一優先度のエッジが存在し、どちらを枝刈りすべきか一意に定まらない場合、リゾルバは推測で動作せず、グラフ解決時に即座にエラーとして失敗します。
-4. **マルチステージ実行 (`AllowMultiStageExecution`) によるデータ保持**:
-   インメモリインベントリタスクなど副作用のない純粋な集約タスクに `AllowMultiStageExecution()` を付与することで、リゾルバはタスクを早期ステージと後期ステージに自動分割・複製します。これにより、高優先度プロデューサのデータを第 1 段階で受け取り、フィードバックプロデューサのデータを第 2 段階で受け取ることで、データを欠落させることなく循環を安全に解消できます。
+   ファンイン依存関係によって循環が検出された場合、グラフリゾルバは優先度順にプロデューサを評価し、サイクルを形成する候補ファンインエッジを決定論的に枝刈り（除外）することで、安全な非循環 DAG を導出します。
 
 ---
 

docs/ja/task-system/01-syntax-and-modes.md

@@ -89,15 +89,13 @@ var ParserTaskA = task.NewTask(
 )
 
 // 3. コンシューマタスクが GetTaskResultsWithTag で全アクティブプロデューサの結果を集約取得
-// 純粋な集約タスクの場合は AllowMultiStageExecution() を指定してマルチステージ実行を許可可能
 var AggregatorTask = task.NewTask(
     AggregatorTaskID,
     []coretask.Dependency{LogItemTag.Ref()},
     func(ctx context.Context) ([]*LogItem, error) {
         items := coretask.GetTaskResultsWithTag(ctx, LogItemTag.Ref())
         return items, nil
     },
-    coretask.AllowMultiStageExecution(),
 )
 ```
 
@@ -110,9 +108,7 @@ var AggregatorTask = task.NewTask(
 2. **Priority による決定論的枝刈りと安定したグラフ**:
    循環を解消するためにエッジを任意に選んで切り落とすと、実行環境やタスク登録順序によって実行順序やデータフローが変動し、再現性のない不安定なグラフになってしまいます。
    - デフォルトで `DefaultTagPriority = 100` が設定される `coretask.WithTagPriority(priority)` により、プロデューサ側がデータの確度や寄与度を宣言します。数値が小さいほど高優先度として扱われます。
-   - グラフリゾルバはサイクル内の最も優先度の低いファンインエッジを決定論的に枝刈りし、常に一意で安定したグラフを導出します。サイクル内の優先度が同着で判断できない場合は推測せず Fail-Fast でエラーを返します。
-3. **マルチステージ実行 (`coretask.AllowMultiStageExecution`)**:
-   エッジを枝刈りしてもデータを欠落させないために、副作用のない純粋な集約タスクには `AllowMultiStageExecution()` を付与します。リゾルバはタスクを分割・複製し、確定的な高優先度プロデューサからの入力をパーサーに渡す早期ステージと、パーサー完了後のフィードバック結果を追加集約する後期ステージの 2 段階で安全に実行します。このラベルのないタスクがマルチステージ実行を要する循環に巻き込まれた場合は、実行時エラーとして検知されます。
+   - グラフリゾルバはサイクルを形成するファンインエッジを決定論的に枝刈り（除外）し、常に安全で一意かつ安定した単一ステージの DAG を導出します。
 
 詳細なアーキテクチャ背景は、[概念ガイド: 6. ファンインにおける循環依存の前提条件と Priority によるグラフ安定化](../khi-task-system-concept.md#6-ファンインにおける循環依存の前提条件と-priority-によるグラフ安定化) を参照してください。
 

pkg/core/inspection/formtask/fileform.go

@@ -22,7 +22,7 @@ import (
 	"github.com/GoogleCloudPlatform/khi/pkg/common/khictx"
 	"github.com/GoogleCloudPlatform/khi/pkg/common/typedmap"
 	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
-	common_task "github.com/GoogleCloudPlatform/khi/pkg/core/task"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	"github.com/GoogleCloudPlatform/khi/pkg/server/upload"
 	core_contract "github.com/GoogleCloudPlatform/khi/pkg/task/core/contract"
@@ -42,7 +42,7 @@ func NewFileFormTaskBuilder(id taskid.TaskImplementationID[upload.UploadResult],
 }
 
 // WithDependencies sets the task dependencies
-func (b *FileFormTaskBuilder) WithDependencies(dependencies []taskid.UntypedTaskReference) *FileFormTaskBuilder {
+func (b *FileFormTaskBuilder) WithDependencies(dependencies []coretask.Dependency) *FileFormTaskBuilder {
 	b.FormTaskBuilderBase.WithDependencies(dependencies)
 	return b
 }
@@ -53,8 +53,8 @@ func (b *FileFormTaskBuilder) WithDescription(description string) *FileFormTaskB
 	return b
 }
 
-func (b *FileFormTaskBuilder) Build(labelOpts ...common_task.LabelOpt) common_task.Task[upload.UploadResult] {
-	return common_task.NewTask(b.FormTaskBuilderBase.id, b.FormTaskBuilderBase.dependencies, func(ctx context.Context) (upload.UploadResult, error) {
+func (b *FileFormTaskBuilder) Build(labelOpts ...coretask.LabelOpt) coretask.Task[upload.UploadResult] {
+	return coretask.NewTask(b.FormTaskBuilderBase.id, b.FormTaskBuilderBase.dependencies, func(ctx context.Context) (upload.UploadResult, error) {
 		metadata := khictx.MustGetValue(ctx, inspectioncore_contract.InspectionRunMetadata)
 
 		req := khictx.MustGetValue(ctx, inspectioncore_contract.InspectionTaskInput)

pkg/core/inspection/formtask/formbase.go

@@ -16,6 +16,7 @@ package formtask
 
 import (
 	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 )
 
@@ -24,7 +25,7 @@ type FormTaskBuilderBase[T any] struct {
 	id           taskid.TaskImplementationID[T]
 	label        string
 	priority     int
-	dependencies []taskid.UntypedTaskReference
+	dependencies []coretask.Dependency
 	description  string
 }
 
@@ -34,7 +35,7 @@ func NewFormTaskBuilderBase[T any](id taskid.TaskImplementationID[T], priority i
 		id:           id,
 		priority:     priority,
 		label:        label,
-		dependencies: []taskid.UntypedTaskReference{},
+		dependencies: []coretask.Dependency{},
 	}
 }
 
@@ -45,7 +46,7 @@ func (b *FormTaskBuilderBase[T]) WithDescription(description string) *FormTaskBu
 }
 
 // WithDependencies sets the task dependencies
-func (b *FormTaskBuilderBase[T]) WithDependencies(dependencies []taskid.UntypedTaskReference) *FormTaskBuilderBase[T] {
+func (b *FormTaskBuilderBase[T]) WithDependencies(dependencies []coretask.Dependency) *FormTaskBuilderBase[T] {
 	b.dependencies = dependencies
 	return b
 }

pkg/core/inspection/formtask/formbase_test.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 )
 
@@ -62,7 +63,7 @@ func TestFormTaskBuilderBase_WithDescription(t *testing.T) {
 
 func TestFormTaskBuilderBase_WithDependencies(t *testing.T) {
 	builder := NewFormTaskBuilderBase(taskid.NewDefaultImplementationID[string]("test-id"), 1, "Test Label")
-	testDependencies := []taskid.UntypedTaskReference{taskid.NewTaskReference[string]("dep1"), taskid.NewTaskReference[string]("dep2")}
+	testDependencies := []coretask.Dependency{taskid.NewTaskReference[string]("dep1"), taskid.NewTaskReference[string]("dep2")}
 
 	result := builder.WithDependencies(testDependencies)
 

pkg/core/inspection/formtask/setform.go

@@ -21,7 +21,7 @@ import (
 	"github.com/GoogleCloudPlatform/khi/pkg/common/khictx"
 	"github.com/GoogleCloudPlatform/khi/pkg/common/typedmap"
 	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
-	common_task "github.com/GoogleCloudPlatform/khi/pkg/core/task"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
 )
@@ -86,7 +86,7 @@ func NewSetFormTaskBuilder[T any](id taskid.TaskImplementationID[T], priority in
 	}
 }
 
-func (b *SetFormTaskBuilder[T]) WithDependencies(dependencies []taskid.UntypedTaskReference) *SetFormTaskBuilder[T] {
+func (b *SetFormTaskBuilder[T]) WithDependencies(dependencies []coretask.Dependency) *SetFormTaskBuilder[T] {
 	b.FormTaskBuilderBase.WithDependencies(dependencies)
 	return b
 }
@@ -183,8 +183,8 @@ func (b *SetFormTaskBuilder[T]) WithConverter(converter SetFormValueConverter[T]
 	return b
 }
 
-func (b *SetFormTaskBuilder[T]) Build(labelOpts ...common_task.LabelOpt) common_task.Task[T] {
-	return common_task.NewTask(b.id, b.dependencies, func(ctx context.Context) (T, error) {
+func (b *SetFormTaskBuilder[T]) Build(labelOpts ...coretask.LabelOpt) coretask.Task[T] {
+	return coretask.NewTask(b.id, b.dependencies, func(ctx context.Context) (T, error) {
 		m := khictx.MustGetValue(ctx, inspectioncore_contract.InspectionRunMetadata)
 		req := khictx.MustGetValue(ctx, inspectioncore_contract.InspectionTaskInput)
 		taskMode := khictx.MustGetValue(ctx, inspectioncore_contract.InspectionTaskMode)