pi-agent: Implement progressive Target detection

2026-07-17 21:01:35 +08:00
parent fbb81ffb40
commit ad651402fd
15 changed files with 558 additions and 31 deletions
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"

 	"proxyctl/internal/profile"
@@ -31,15 +32,41 @@ type Intent struct {
 	Enabled  bool
 }

+type TargetDetectionPhase uint8
+
+const (
+	// TargetReady is the zero value so snapshots supplied by non-progressive
+	// Modules remain compatible with the established dashboard contract.
+	TargetReady TargetDetectionPhase = iota
+	TargetChecking
+	TargetDetectionFailed
+)
+
 type TargetSnapshot struct {
 	ID                   string
 	Description          target.Description
+	DetectionPhase       TargetDetectionPhase
+	DetectionErr         error
 	Available            bool
 	AvailabilityEvidence string
 	State                target.State
 	Evidence             string
 }

+type TargetDetectionResult struct {
+	TargetID     string
+	Phase        TargetDetectionPhase
+	Availability target.Availability
+	Inspection   target.Inspection
+	Err          error
+}
+
+type TargetDetectionUpdate struct {
+	Snapshot Snapshot
+	Result   TargetDetectionResult
+	Complete bool
+}
+
 type Snapshot struct {
 	Profiles      []profile.Profile
 	ActiveProfile string
@@ -75,12 +102,22 @@ type ProgressiveModule interface {
 	DispatchProgress(Intent) <-chan Result
 }

+// ProgressiveDetectionModule exposes a useful all-Checking snapshot
+// synchronously, followed by one structured Target result at a time. Startup
+// and explicit refresh use this same presentation-independent contract.
+type ProgressiveDetectionModule interface {
+	BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate)
+}
+
 type Application struct {
 	store          profile.Store
 	state          profile.State
 	targets        []target.Target
 	availabilities map[string]target.Availability
 	inspections    map[string]target.Inspection
+	detectionPhase map[string]TargetDetectionPhase
+	detectionErr   map[string]error
+	detectionMu    sync.RWMutex
 	mutationLock   MutationLock
 	logger         OperationLogger
 }
@@ -112,24 +149,93 @@ func NewWithTargetsAndRuntime(store profile.Store, targets []target.Target, muta
 		targets:        append([]target.Target(nil), targets...),
 		availabilities: make(map[string]target.Availability),
 		inspections:    make(map[string]target.Inspection),
+		detectionPhase: make(map[string]TargetDetectionPhase),
+		detectionErr:   make(map[string]error),
 		mutationLock:   mutationLock,
 		logger:         logger,
 	}
-	a.refreshTargets()
+	for _, adapter := range a.targets {
+		a.detectionPhase[adapter.ID()] = TargetChecking
+	}
 	return a, nil
 }

 func (a *Application) Snapshot() Snapshot {
+	a.detectionMu.RLock()
+	defer a.detectionMu.RUnlock()
+	return a.snapshotLocked()
+}
+
+func (a *Application) snapshotLocked() Snapshot {
 	profiles := append([]profile.Profile(nil), a.state.Profiles...)
 	snapshot := Snapshot{Profiles: profiles, ActiveProfile: a.state.ActiveProfile}
 	for _, adapter := range a.targets {
-		availability := a.availabilities[adapter.ID()]
-		inspection := a.inspections[adapter.ID()]
-		snapshot.Targets = append(snapshot.Targets, TargetSnapshot{ID: adapter.ID(), Description: adapter.Description(), Available: availability.Available, AvailabilityEvidence: availability.Evidence, State: inspection.State, Evidence: inspection.Evidence})
+		id := adapter.ID()
+		availability := a.availabilities[id]
+		inspection := a.inspections[id]
+		snapshot.Targets = append(snapshot.Targets, TargetSnapshot{
+			ID: id, Description: adapter.Description(), DetectionPhase: a.detectionPhase[id], DetectionErr: a.detectionErr[id],
+			Available: availability.Available, AvailabilityEvidence: availability.Evidence, State: inspection.State, Evidence: inspection.Evidence,
+		})
 	}
 	return snapshot
 }

+func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate) {
+	a.detectionMu.Lock()
+	for _, adapter := range a.targets {
+		id := adapter.ID()
+		a.detectionPhase[id] = TargetChecking
+		a.detectionErr[id] = nil
+	}
+	checking := a.snapshotLocked()
+	a.detectionMu.Unlock()
+
+	updates := make(chan TargetDetectionUpdate, len(a.targets)+1)
+	active := a.activeProfile()
+	outcomes := make(chan TargetDetectionResult, len(a.targets))
+	for _, adapter := range a.targets {
+		adapter := adapter
+		go func() {
+			detection := detectTarget(adapter, active)
+			phase := TargetReady
+			if detection.Err != nil {
+				phase = TargetDetectionFailed
+			}
+			outcomes <- TargetDetectionResult{
+				TargetID: adapter.ID(), Phase: phase, Availability: detection.Availability,
+				Inspection: detection.Inspection, Err: detection.Err,
+			}
+		}()
+	}
+	go func() {
+		defer close(updates)
+		for range a.targets {
+			result := <-outcomes
+			a.detectionMu.Lock()
+			a.availabilities[result.TargetID] = result.Availability
+			a.inspections[result.TargetID] = result.Inspection
+			a.detectionPhase[result.TargetID] = result.Phase
+			a.detectionErr[result.TargetID] = result.Err
+			snapshot := a.snapshotLocked()
+			a.detectionMu.Unlock()
+			updates <- TargetDetectionUpdate{Snapshot: snapshot, Result: result}
+		}
+		updates <- TargetDetectionUpdate{Snapshot: a.Snapshot(), Complete: true}
+	}()
+	return checking, updates
+}
+
+func detectTarget(adapter target.Target, active *profile.Profile) target.Detection {
+	if detector, ok := adapter.(target.Detector); ok {
+		return detector.Detect(context.Background(), active)
+	}
+	return target.Detection{
+		Availability: adapter.Availability(context.Background()),
+		Inspection:   adapter.Inspect(context.Background(), active),
+	}
+}
+
 func (a *Application) Dispatch(intent Intent) Result {
 	return a.dispatch(intent, nil)
 }
@@ -319,17 +425,18 @@ func (a *Application) activeProfile() *profile.Profile {
 }

 func (a *Application) refreshTargets() {
-	active := a.activeProfile()
-	for _, adapter := range a.targets {
-		a.availabilities[adapter.ID()] = adapter.Availability(context.Background())
-		a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
+	_, updates := a.BeginTargetDetection()
+	for range updates {
 	}
 }

 func (a *Application) refreshInspections() {
 	active := a.activeProfile()
 	for _, adapter := range a.targets {
-		a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
+		inspection := adapter.Inspect(context.Background(), active)
+		a.detectionMu.Lock()
+		a.inspections[adapter.ID()] = inspection
+		a.detectionMu.Unlock()
 	}
 }

@@ -0,0 +1,127 @@
+package app_test
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"reflect"
+	"testing"
+
+	"proxyctl/internal/app"
+	"proxyctl/internal/profile"
+	"proxyctl/internal/target"
+)
+
+type controlledDetectionTarget struct {
+	id          string
+	description target.Description
+	release     <-chan struct{}
+	detection   target.Detection
+}
+
+func (t *controlledDetectionTarget) ID() string                      { return t.id }
+func (t *controlledDetectionTarget) Description() target.Description { return t.description }
+func (t *controlledDetectionTarget) Availability(context.Context) target.Availability {
+	panic("progressive detection must use the structured Detector seam")
+}
+func (t *controlledDetectionTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
+	panic("progressive detection must use the structured Detector seam")
+}
+func (t *controlledDetectionTarget) Confirmation(bool) string { return "" }
+func (t *controlledDetectionTarget) SetDesired(context.Context, bool, *profile.Profile) target.OperationResult {
+	return target.OperationResult{}
+}
+func (t *controlledDetectionTarget) Detect(context.Context, *profile.Profile) target.Detection {
+	<-t.release
+	return t.detection
+}
+
+func TestApplicationStreamsIndependentTargetDetectionInProductOrder(t *testing.T) {
+	firstRelease := make(chan struct{})
+	secondRelease := make(chan struct{})
+	failedRelease := make(chan struct{})
+	adapters := []target.Target{
+		&controlledDetectionTarget{id: "first", description: target.Description{Name: "First", Summary: "first description"}, release: firstRelease, detection: target.Detection{Availability: target.Availability{Available: true, Evidence: "first available"}, Inspection: target.Inspection{State: target.Enabled, Evidence: "first enabled"}}},
+		&controlledDetectionTarget{id: "second", description: target.Description{Name: "Second", Summary: "second description"}, release: secondRelease, detection: target.Detection{Availability: target.Availability{Available: true, Evidence: "second available"}, Inspection: target.Inspection{State: target.External, Evidence: "second external"}}},
+		&controlledDetectionTarget{id: "failed", description: target.Description{Name: "Failed", Summary: "failed description"}, release: failedRelease, detection: target.Detection{Err: errors.New("inspection command failed")}},
+	}
+	application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), adapters)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	before := application.Snapshot()
+	assertTargetOrder(t, before, []string{"first", "second", "failed"})
+	for _, item := range before.Targets {
+		if item.DetectionPhase != app.TargetChecking || item.Description.Name == "" || item.Description.Summary == "" {
+			t.Fatalf("Target identity was not useful before detection: %#v", item)
+		}
+	}
+
+	checking, updates := application.BeginTargetDetection()
+	for _, item := range checking.Targets {
+		if item.DetectionPhase != app.TargetChecking {
+			t.Fatalf("Target did not begin in Checking: %#v", item)
+		}
+	}
+
+	close(secondRelease)
+	second := <-updates
+	if second.Result.TargetID != "second" || second.Result.Phase != app.TargetReady || second.Result.Inspection.State != target.External {
+		t.Fatalf("second completion = %#v", second)
+	}
+	assertOnlyDetectionChanged(t, checking, second.Snapshot, "second")
+	assertTargetOrder(t, second.Snapshot, []string{"first", "second", "failed"})
+
+	close(failedRelease)
+	failed := <-updates
+	if failed.Result.TargetID != "failed" || failed.Result.Phase != app.TargetDetectionFailed || failed.Result.Err == nil {
+		t.Fatalf("failed completion = %#v", failed)
+	}
+	assertOnlyDetectionChanged(t, second.Snapshot, failed.Snapshot, "failed")
+
+	close(firstRelease)
+	first := <-updates
+	if first.Result.TargetID != "first" || first.Result.Phase != app.TargetReady || first.Result.Inspection.State != target.Enabled {
+		t.Fatalf("first completion = %#v", first)
+	}
+	assertOnlyDetectionChanged(t, failed.Snapshot, first.Snapshot, "first")
+
+	final := <-updates
+	if !final.Complete {
+		t.Fatalf("final update = %#v", final)
+	}
+	assertTargetOrder(t, final.Snapshot, []string{"first", "second", "failed"})
+	if final.Snapshot.Targets[0].State != target.Enabled || final.Snapshot.Targets[1].State != target.External || final.Snapshot.Targets[2].DetectionPhase != app.TargetDetectionFailed {
+		t.Fatalf("final snapshot = %#v", final.Snapshot.Targets)
+	}
+}
+
+func assertTargetOrder(t *testing.T, snapshot app.Snapshot, want []string) {
+	t.Helper()
+	got := make([]string, len(snapshot.Targets))
+	for i := range snapshot.Targets {
+		got[i] = snapshot.Targets[i].ID
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("Target order = %#v, want %#v", got, want)
+	}
+}
+
+func assertOnlyDetectionChanged(t *testing.T, before, after app.Snapshot, changedID string) {
+	t.Helper()
+	if len(before.Targets) != len(after.Targets) {
+		t.Fatalf("Target count changed from %d to %d", len(before.Targets), len(after.Targets))
+	}
+	for i := range before.Targets {
+		if before.Targets[i].ID == changedID {
+			if reflect.DeepEqual(before.Targets[i], after.Targets[i]) {
+				t.Fatalf("completed Target %q did not change", changedID)
+			}
+			continue
+		}
+		if !reflect.DeepEqual(before.Targets[i], after.Targets[i]) {
+			t.Fatalf("completion for %q changed unrelated Target %q: before=%#v after=%#v", changedID, before.Targets[i].ID, before.Targets[i], after.Targets[i])
+		}
+	}
+}
@@ -49,6 +49,9 @@ func TestApplicationTargetSeamRequiresActiveProfileAndReturnsTruthfulResults(t *
 	if err != nil {
 		t.Fatal(err)
 	}
+	if result := application.Dispatch(app.Intent{Kind: app.RefreshTargets}); result.Err != nil {
+		t.Fatal(result.Err)
+	}
 	initial := application.Snapshot()
 	if len(initial.Targets) != 1 || initial.Targets[0].State != target.External || initial.Targets[0].Description.Summary != "Shell summary" {
 		t.Fatalf("initial Target snapshot = %#v", initial.Targets)
@@ -104,6 +107,9 @@ func TestApplicationHidesUnavailableTargetButKeepsExplanation(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	if result := application.Dispatch(app.Intent{Kind: app.RefreshTargets}); result.Err != nil {
+		t.Fatal(result.Err)
+	}
 	got := application.Snapshot().Targets[0]
 	if got.Available || got.AvailabilityEvidence != "fake detection" {
 		t.Fatalf("unavailable Target = %#v", got)