gitea-admin/proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pi-agent: Fix final integrated review

Browse Source
This commit is contained in:
proxyctl coordinator
2026-07-18 02:37:54 +08:00
parent ed8b527f1f
commit 652d7eba2b
18 changed files with 566 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/app.go
+2 -21
View File
@@ -19,7 +19,6 @@ const (
UpdateProfile
DeleteProfile
ActivateProfile
SetTargetDesired // retained for synchronous compatibility; interactive callers use the explicit intents below.
RefreshTargets
EnableTarget
DisableTarget
@@ -34,7 +33,6 @@ type Intent struct {
Profile profile.Profile
Confirm bool
TargetID string
Enabled bool
}
// TargetActionKind identifies a Target action independently of display labels.
@@ -726,9 +724,6 @@ func (a *Application) refreshInspections() {
}
func isTargetMutationIntent(kind IntentKind) bool {
if kind == SetTargetDesired {
return true
}
for _, policy := range targetActionPolicies {
if policy.intent == kind {
return true
@@ -737,10 +732,7 @@ func isTargetMutationIntent(kind IntentKind) bool {
return false
}
func requestedTargetAction(intent Intent, selected TargetSnapshot) (TargetActionKind, bool) {
if intent.Kind == SetTargetDesired {
return legacyTargetAction(intent.Enabled, selected.State)
}
func requestedTargetAction(intent Intent) (TargetActionKind, bool) {
for _, policy := range targetActionPolicies {
if policy.intent == intent.Kind {
return policy.kind, true
@@ -749,17 +741,6 @@ func requestedTargetAction(intent Intent, selected TargetSnapshot) (TargetAction
return 0, false
}
// legacyTargetAction is the only compatibility boundary for the former
// SetTargetDesired boolean contract. All current callers use TargetActionKind.
func legacyTargetAction(enabled bool, state target.State) (TargetActionKind, bool) {
for _, policy := range targetActionPolicies {
if policy.intent != 0 && policy.state == state && policy.desired == enabled {
return policy.kind, true
}
}
return 0, false
}
func structuredTargetAction(selected TargetSnapshot, kind TargetActionKind) (TargetAction, bool) {
if selected.PrimaryAction.Kind == kind {
return selected.PrimaryAction, true
@@ -773,7 +754,7 @@ func structuredTargetAction(selected TargetSnapshot, kind TargetActionKind) (Tar
}
func validateTargetAction(intent Intent, selected TargetSnapshot) (targetActionPolicy, error) {
kind, ok := requestedTargetAction(intent, selected)
kind, ok := requestedTargetAction(intent)
if !ok {
return targetActionPolicy{}, fmt.Errorf("Target %q action is no longer available because its state is %s", selected.ID, selected.State)
}
internal/app/coordination_test.go
+8 -8
View File
@@ -92,7 +92,7 @@ func TestSecondApplicationUsesPersistedActiveProfileChangedByFirstApplication(t
if result := first.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "new"}); result.Err != nil {
t.Fatalf("first Application activates new Profile: %v", result.Err)
}
result := second.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
result := second.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
if result.Err != nil {
t.Fatalf("second Application enables Target: %v", result.Err)
}
@@ -122,7 +122,7 @@ func TestActiveProfileSwitchCoordinatesEnabledTargetsWithoutRollback(t *testing.
}
refreshApplicationTargets(t, application)
for _, id := range []string{"first", "second"} {
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: id, Enabled: true}); result.Err != nil {
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: id}); result.Err != nil {
t.Fatal(result.Err)
}
}
@@ -192,7 +192,7 @@ func TestApplicationCoordinatesMultipleRealAndFakeHermeticAdaptersThroughPartial
}
refreshApplicationTargets(t, application)
for _, id := range []string{shellAdapter.ID(), dockerAdapter.ID(), failure.ID()} {
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: id, Enabled: true}); result.Err != nil {
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: id}); result.Err != nil {
t.Fatalf("enable %s: %v", id, result.Err)
}
}
@@ -233,7 +233,7 @@ func TestActiveProfileProgressReusesAvailabilityCache(t *testing.T) {
t.Fatal(result.Err)
}
refreshApplicationTargets(t, application)
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true}); result.Err != nil {
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"}); result.Err != nil {
t.Fatal(result.Err)
}
@@ -269,7 +269,7 @@ func TestActiveProfileSwitchStreamsApplyingAndCompletionForEachTarget(t *testing
t.Fatal(result.Err)
}
refreshApplicationTargets(t, application)
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true}); result.Err != nil {
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"}); result.Err != nil {
t.Fatal(result.Err)
}
@@ -315,7 +315,7 @@ func TestDeletingFinalProfileLeavesManagedTargetsDriftedAndDisableStillRestoresE
t.Fatal(activated.Err)
}
refreshApplicationTargets(t, application)
if enabled := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true}); enabled.Err != nil {
if enabled := application.Dispatch(app.Intent{Kind: app.ApplyProfileToTarget, TargetID: adapter.ID()}); enabled.Err != nil {
t.Fatal(enabled.Err)
}
@@ -327,11 +327,11 @@ func TestDeletingFinalProfileLeavesManagedTargetsDriftedAndDisableStillRestoresE
if deleted.Err != nil || deleted.Snapshot.ActiveProfile != "" || deleted.Snapshot.Targets[0].State != target.Drifted {
t.Fatalf("confirmed final delete = %#v", deleted)
}
enable := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
enable := application.Dispatch(app.Intent{Kind: app.ReapplyActiveProfile, TargetID: adapter.ID()})
if enable.Err == nil || !strings.Contains(enable.Err.Error(), "Profile management") {
t.Fatalf("re-enable without Active Profile = %v", enable.Err)
}
disabled := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: false})
disabled := application.Dispatch(app.Intent{Kind: app.RestoreOriginal, TargetID: adapter.ID()})
if disabled.Err != nil || disabled.Snapshot.Targets[0].State != target.External {
t.Fatalf("disable after final delete = %#v", disabled)
}
internal/app/maintenance_test.go
+1 -1
View File
@@ -127,7 +127,7 @@ func TestApplicationSerializesAndLogsTargetMutations(t *testing.T) {
adapter.inspection = target.Inspection{State: target.Enabled}
refreshApplicationTargets(t, application)
adapter.disableErr = errors.New("failed http://alice:secret@proxy.test/private")
result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "fake", Enabled: false})
result := application.Dispatch(app.Intent{Kind: app.DisableTarget, TargetID: "fake"})
if result.Err == nil {
t.Fatal("Target failure reported success")
}
internal/app/operation.go
+63 -70
View File
@@ -42,6 +42,34 @@ const (
// exactly one of running, confirmation, succeeded, failed, or closed.
type OperationState interface {
operationState()
decode() DecodedOperationUpdate
}
// OperationUpdateKind is the decoded variant of one operation stream update.
// Consumers decode once and handle the payload without repeating type switches.
type OperationUpdateKind uint8
const (
OperationRunning OperationUpdateKind = iota + 1
OperationConfirmation
OperationSucceeded
OperationFailed
OperationClosed
)
// DecodedOperationUpdate is the complete behavior-bearing projection of an
// OperationUpdate variant. Fields that do not apply to Kind retain zero values.
type DecodedOperationUpdate struct {
Kind OperationUpdateKind
Status OperationStatus
Progress []TargetProgress
Confirmation *Confirmation
Err error
FirstFailedTargetID string
}
func (u DecodedOperationUpdate) IsTerminal() bool {
return u.Kind != OperationRunning
}
type operationRunning struct {
@@ -50,10 +78,18 @@ type operationRunning struct {
}
func (operationRunning) operationState() {}
func (state operationRunning) decode() DecodedOperationUpdate {
return DecodedOperationUpdate{Kind: OperationRunning, Status: state.status, Progress: cloneProgress(state.progress)}
}
type operationConfirmation struct{ request Confirmation }
func (operationConfirmation) operationState() {}
func (state operationConfirmation) decode() DecodedOperationUpdate {
request := state.request
request.Details = append([]string(nil), request.Details...)
return DecodedOperationUpdate{Kind: OperationConfirmation, Confirmation: &request}
}
type operationSucceeded struct {
status OperationStatus
@@ -61,6 +97,9 @@ type operationSucceeded struct {
}
func (operationSucceeded) operationState() {}
func (state operationSucceeded) decode() DecodedOperationUpdate {
return DecodedOperationUpdate{Kind: OperationSucceeded, Status: state.status, Progress: cloneProgress(state.progress)}
}
type operationFailed struct {
status OperationStatus
@@ -70,10 +109,19 @@ type operationFailed struct {
}
func (operationFailed) operationState() {}
func (state operationFailed) decode() DecodedOperationUpdate {
return DecodedOperationUpdate{
Kind: OperationFailed, Status: state.status, Progress: cloneProgress(state.progress),
Err: state.err, FirstFailedTargetID: state.firstFailedTargetID,
}
}
type operationClosed struct{ status OperationStatus }
func (operationClosed) operationState() {}
func (state operationClosed) decode() DecodedOperationUpdate {
return DecodedOperationUpdate{Kind: OperationClosed, Status: state.status}
}
// OperationUpdate is the common progressive contract used by interactive
// detection, Target mutation, and Active Profile transition operations. Build
@@ -116,79 +164,24 @@ func cloneProgress(progress []TargetProgress) []TargetProgress {
return append([]TargetProgress(nil), progress...)
}
func (u OperationUpdate) Status() OperationStatus {
switch state := u.State.(type) {
case operationRunning:
return state.status
case operationSucceeded:
return state.status
case operationFailed:
return state.status
case operationClosed:
return state.status
default:
return OperationStatus{}
// Decode centralizes the closed-set state decoding for all consumers.
func (u OperationUpdate) Decode() DecodedOperationUpdate {
if u.State == nil {
return DecodedOperationUpdate{}
}
return u.State.decode()
}
func (u OperationUpdate) Progress() []TargetProgress {
switch state := u.State.(type) {
case operationRunning:
return cloneProgress(state.progress)
case operationSucceeded:
return cloneProgress(state.progress)
case operationFailed:
return cloneProgress(state.progress)
default:
return nil
}
}
func (u OperationUpdate) Confirmation() *Confirmation {
state, ok := u.State.(operationConfirmation)
if !ok {
return nil
}
request := state.request
request.Details = append([]string(nil), request.Details...)
return &request
}
func (u OperationUpdate) Err() error {
if state, ok := u.State.(operationFailed); ok {
return state.err
}
return nil
}
func (u OperationUpdate) FirstFailedTargetID() string {
if state, ok := u.State.(operationFailed); ok {
return state.firstFailedTargetID
}
return ""
}
func (u OperationUpdate) IsRunning() bool {
_, ok := u.State.(operationRunning)
return ok
}
func (u OperationUpdate) IsConfirmation() bool {
_, ok := u.State.(operationConfirmation)
return ok
}
func (u OperationUpdate) IsSucceeded() bool {
_, ok := u.State.(operationSucceeded)
return ok
}
func (u OperationUpdate) IsFailed() bool {
_, ok := u.State.(operationFailed)
return ok
}
func (u OperationUpdate) IsTerminal() bool { return !u.IsRunning() }
func (u OperationUpdate) Status() OperationStatus { return u.Decode().Status }
func (u OperationUpdate) Progress() []TargetProgress { return u.Decode().Progress }
func (u OperationUpdate) Confirmation() *Confirmation { return u.Decode().Confirmation }
func (u OperationUpdate) Err() error { return u.Decode().Err }
func (u OperationUpdate) FirstFailedTargetID() string { return u.Decode().FirstFailedTargetID }
func (u OperationUpdate) IsRunning() bool { return u.Decode().Kind == OperationRunning }
func (u OperationUpdate) IsConfirmation() bool { return u.Decode().Kind == OperationConfirmation }
func (u OperationUpdate) IsSucceeded() bool { return u.Decode().Kind == OperationSucceeded }
func (u OperationUpdate) IsFailed() bool { return u.Decode().Kind == OperationFailed }
func (u OperationUpdate) IsTerminal() bool { return u.Decode().IsTerminal() }
// OperationStream exposes the useful initial frame synchronously and all
// subsequent structured changes through Updates. Updates closes after a
internal/app/operation_stream_test.go
+6 -6
View File
@@ -97,7 +97,7 @@ func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgr
}
refreshApplicationTargets(t, application)
operation := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
running := <-operation.Updates
runningProgress := running.Progress()
if !running.IsRunning() || len(runningProgress) != 1 || runningProgress[0].Phase != app.TargetProgressRunning || runningProgress[0].TargetID != "shell" || running.Status().Kind != app.StatusProgress {
@@ -105,7 +105,7 @@ func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgr
}
<-adapter.started
blocked := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: false})
blocked := application.StartOperation(app.Intent{Kind: app.DisableTarget, TargetID: "shell"})
blockedUpdate := <-blocked.Updates
if !blockedUpdate.IsTerminal() || blockedUpdate.Status().Kind != app.StatusBlocked || !strings.Contains(blockedUpdate.Status().Summary, "already in progress") {
t.Fatalf("blocked update = %#v", blockedUpdate)
@@ -148,7 +148,7 @@ func TestDetectionOperationSerializesWithSynchronousMutation(t *testing.T) {
mutationDone := make(chan app.Result, 1)
go func() {
close(mutationAttempted)
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
mutationDone <- application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
}()
<-mutationAttempted
close(adapter.detectionRelease)
@@ -182,7 +182,7 @@ func TestStartOperationReturnsBeforeDispatchSerializationIsAvailable(t *testing.
mutationDone := make(chan app.Result, 1)
go func() {
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
mutationDone <- application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
}()
<-adapter.started // barrier: synchronous Dispatch owns dispatch serialization
adapter.detectionStarted = make(chan struct{})
@@ -221,7 +221,7 @@ func TestInteractiveOperationStreamUsesStructuredConfirmation(t *testing.T) {
}
refreshApplicationTargets(t, application)
operation := application.StartOperation(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
update := <-operation.Updates
confirmation := update.Confirmation()
if !update.IsConfirmation() || confirmation == nil || confirmation.Title == "" || confirmation.CancelLabel == "" {
@@ -249,7 +249,7 @@ func TestInteractiveProfileTransitionReportsFirstFailureWithoutRollback(t *testi
}
refreshApplicationTargets(t, application)
for _, id := range []string{"first", "second"} {
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: id, Enabled: true}); result.Err != nil {
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: id}); result.Err != nil {
t.Fatal(result.Err)
}
}
internal/app/progressive_detection_test.go
+1 -1
View File
@@ -220,7 +220,7 @@ func TestDispatchMutationWaitsForInFlightTargetDetection(t *testing.T) {
<-adapter.started
mutationDone := make(chan app.Result, 1)
go func() {
mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
mutationDone <- application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
}()
close(adapter.release)
for range detection.Updates {
internal/app/target_test.go
+3 -3
View File
@@ -60,20 +60,20 @@ func TestApplicationTargetSeamRequiresActiveProfileAndReturnsTruthfulResults(t *
t.Fatalf("initial Target snapshot = %#v", initial.Targets)
}
result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
result := application.Dispatch(app.Intent{Kind: app.ApplyProfileToTarget, TargetID: "shell"})
if result.Err == nil || !strings.Contains(result.Err.Error(), "Profile management") || len(adapter.calls) != 0 || result.Snapshot.Targets[0].State != target.External {
t.Fatalf("enable without Active Profile = %#v, calls=%v", result, adapter.calls)
}
createAndActivate(t, application)
result = application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true})
result = application.Dispatch(app.Intent{Kind: app.ApplyProfileToTarget, TargetID: "shell"})
if result.Err != nil || result.Snapshot.Targets[0].State != target.Enabled || len(adapter.calls) != 1 {
t.Fatalf("enable result = %#v", result)
}
adapter.setErr = errors.New("permission denied")
adapter.state = target.Drifted
result = application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: false})
result = application.Dispatch(app.Intent{Kind: app.RestoreOriginal, TargetID: "shell"})
if result.Err == nil || result.Snapshot.Targets[0].State != target.Drifted || result.Snapshot.Targets[0].Evidence != "actual state" {
t.Fatalf("failed operation not truthful: %#v", result)
}