fix(workflow): stop agent profile duplication from rebinding synced steps (#2740)
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -0,0 +1,247 @@
|
||||
package backendapp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jmoiron/sqlx"
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
|
||||
agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models"
|
||||
settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
// newMatcherTestRepos opens an isolated SQLite-backed agent settings store for
|
||||
// buildAgentProfileMatcher tests, along with the raw *sqlx.DB handle so a test
|
||||
// can force specific CreatedAt values that the store API always overwrites.
|
||||
func newMatcherTestRepos(t *testing.T) (*Repositories, *sqlx.DB) {
|
||||
t.Helper()
|
||||
db, err := sqlx.Open("sqlite3", filepath.Join(t.TempDir(), "matcher.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open sqlite: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"})
|
||||
if err != nil {
|
||||
t.Fatalf("new logger: %v", err)
|
||||
}
|
||||
repo, cleanup, err := settingsstore.Provide(db, db, log)
|
||||
if err != nil {
|
||||
t.Fatalf("settings store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = cleanup() })
|
||||
return &Repositories{AgentSettings: repo}, db
|
||||
}
|
||||
|
||||
func createMatcherTestAgent(t *testing.T, repos *Repositories) string {
|
||||
t.Helper()
|
||||
agent := &agentsettingsmodels.Agent{Name: "claude-" + uuid.New().String()}
|
||||
if err := repos.AgentSettings.CreateAgent(context.Background(), agent); err != nil {
|
||||
t.Fatalf("CreateAgent: %v", err)
|
||||
}
|
||||
return agent.ID
|
||||
}
|
||||
|
||||
// createMatcherTestProfile creates a profile with the given (agent display
|
||||
// name, model, mode) triple, applying any overrides after creation.
|
||||
func createMatcherTestProfile(t *testing.T, repos *Repositories, agentID, displayName, model, mode string) *agentsettingsmodels.AgentProfile {
|
||||
t.Helper()
|
||||
p := &agentsettingsmodels.AgentProfile{
|
||||
AgentID: agentID,
|
||||
Name: "Profile " + uuid.New().String(),
|
||||
AgentDisplayName: displayName,
|
||||
Model: model,
|
||||
Mode: mode,
|
||||
}
|
||||
if err := repos.AgentSettings.CreateAgentProfile(context.Background(), p); err != nil {
|
||||
t.Fatalf("CreateAgentProfile: %v", err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// TestBuildAgentProfileMatcher_DuplicateDoesNotStealBinding is the reported
|
||||
// regression: duplicating a profile produces a byte-identical
|
||||
// (agent_display_name, model, mode) triple, and the copy's CreatedAt is
|
||||
// always later than its source. The matcher must keep resolving to the
|
||||
// original (oldest) profile, not the newer copy.
|
||||
func TestBuildAgentProfileMatcher_DuplicateDoesNotStealBinding(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
source := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
copyProfile := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto") // the "Copy"
|
||||
require.True(t, source.CreatedAt.Before(copyProfile.CreatedAt),
|
||||
"precondition: the copy must be strictly newer than the source, or this test stops proving the regression")
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != source.ID {
|
||||
t.Fatalf("matcher returned %q, want source profile %q", got, source.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAgentProfileMatcher_SkipsDisabledProfile(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
p := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
if _, err := repos.AgentSettings.UpdateAgentProfileEnabled(context.Background(), p.ID, false); err != nil {
|
||||
t.Fatalf("disable profile: %v", err)
|
||||
}
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != "" {
|
||||
t.Fatalf("matcher returned %q, want empty (disabled profile must never match)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildAgentProfileMatcher_PreservesDisabledExistingBinding is the
|
||||
// reconciliation regression: a workflow step already bound to a profile that
|
||||
// gets disabled after the fact must keep that binding on the next sync, per
|
||||
// docs/specs/agents/profile-disable.md ("existing sessions/workflows keep
|
||||
// their profile"). Excluding disabled profiles from *candidate selection*
|
||||
// must not also strip a binding that was already in place.
|
||||
func TestBuildAgentProfileMatcher_PreservesDisabledExistingBinding(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
p := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
if _, err := repos.AgentSettings.UpdateAgentProfileEnabled(context.Background(), p.ID, false); err != nil {
|
||||
t.Fatalf("disable profile: %v", err)
|
||||
}
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", p.ID)
|
||||
if got != p.ID {
|
||||
t.Fatalf("matcher returned %q, want existing binding %q preserved despite disablement", got, p.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildAgentProfileMatcher_CurrentIDDriftFallsBackToCandidateSelection
|
||||
// covers the case where the currently-bound profile no longer matches the
|
||||
// descriptor (e.g. the profile itself was edited) - the matcher must not
|
||||
// blindly trust currentID, it must fall through to normal candidate
|
||||
// selection.
|
||||
func TestBuildAgentProfileMatcher_CurrentIDDriftFallsBackToCandidateSelection(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
stale := createMatcherTestProfile(t, repos, agentID, "Claude", "sonnet[1m]", "auto")
|
||||
fresh := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", stale.ID)
|
||||
if got != fresh.ID {
|
||||
t.Fatalf("matcher returned %q, want %q (currentID no longer matches the descriptor)", got, fresh.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAgentProfileMatcher_SkipsWorkspaceScopedProfile(t *testing.T) {
|
||||
repos, db := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
p := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
if _, err := db.Exec(`UPDATE agent_profiles SET workspace_id = ? WHERE id = ?`, "workspace-1", p.ID); err != nil {
|
||||
t.Fatalf("set workspace_id: %v", err)
|
||||
}
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != "" {
|
||||
t.Fatalf("matcher returned %q, want empty (workspace-scoped profile must never match)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAgentProfileMatcher_SingleMatchUnchanged(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
p := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != p.ID {
|
||||
t.Fatalf("matcher returned %q, want %q", got, p.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAgentProfileMatcher_NoMatch(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
got := match("Claude", "sonnet[1m]", "auto", "")
|
||||
if got != "" {
|
||||
t.Fatalf("matcher returned %q, want empty", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildAgentProfileMatcher_EqualCreatedAtIsStable forces two candidates
|
||||
// to share an identical CreatedAt (which the store API's time.Now() calls
|
||||
// would otherwise make astronomically unlikely) and asserts the ID tiebreak
|
||||
// makes the result deterministic across repeated calls.
|
||||
func TestBuildAgentProfileMatcher_EqualCreatedAtIsStable(t *testing.T) {
|
||||
repos, db := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
a := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
b := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
if _, err := db.Exec(`UPDATE agent_profiles SET created_at = (SELECT created_at FROM agent_profiles WHERE id = ?) WHERE id = ?`, a.ID, b.ID); err != nil {
|
||||
t.Fatalf("force equal created_at: %v", err)
|
||||
}
|
||||
|
||||
want := a.ID
|
||||
if b.ID < a.ID {
|
||||
want = b.ID
|
||||
}
|
||||
|
||||
match := buildAgentProfileMatcher(repos, testLogger(t))
|
||||
for i := 0; i < 5; i++ {
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != want {
|
||||
t.Fatalf("call %d: matcher returned %q, want stable %q", i, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildAgentProfileMatcher_MultipleCandidatesLogsDebugWithFields proves
|
||||
// the ambiguity signal described in buildAgentProfileMatcher's doc comment:
|
||||
// when more than one candidate matches, it logs at Debug with the triple,
|
||||
// the candidate count, and the profile it actually selected.
|
||||
func TestBuildAgentProfileMatcher_MultipleCandidatesLogsDebugWithFields(t *testing.T) {
|
||||
repos, _ := newMatcherTestRepos(t)
|
||||
agentID := createMatcherTestAgent(t, repos)
|
||||
|
||||
a := createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto")
|
||||
createMatcherTestProfile(t, repos, agentID, "Claude", "opus[1m]", "auto") // newer candidate, must not be selected
|
||||
|
||||
core, logs := observer.New(zapcore.DebugLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
match := buildAgentProfileMatcher(repos, log)
|
||||
got := match("Claude", "opus[1m]", "auto", "")
|
||||
if got != a.ID {
|
||||
t.Fatalf("matcher returned %q, want oldest profile %q", got, a.ID)
|
||||
}
|
||||
|
||||
entries := logs.FilterMessage("agent profile matcher: multiple candidates for workflow step sync").All()
|
||||
require.Len(t, entries, 1, "an ambiguous match must log exactly one Debug entry")
|
||||
fields := entries[0].ContextMap()
|
||||
assert.Equal(t, "Claude", fields["agent_display_name"])
|
||||
assert.Equal(t, "opus[1m]", fields["model"])
|
||||
assert.Equal(t, "auto", fields["mode"])
|
||||
assert.Equal(t, int64(2), fields["candidates"])
|
||||
assert.Equal(t, a.ID, fields["selected_profile_id"])
|
||||
}
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/agent/managedruntime"
|
||||
"github.com/kandev/kandev/internal/agent/registry"
|
||||
agentsettingscontroller "github.com/kandev/kandev/internal/agent/settings/controller"
|
||||
agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models"
|
||||
analyticsservice "github.com/kandev/kandev/internal/analytics/service"
|
||||
"github.com/kandev/kandev/internal/automation"
|
||||
"github.com/kandev/kandev/internal/azuredevops"
|
||||
@@ -153,7 +154,7 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories
|
||||
// Wire agent profile resolver/matcher for workflow export/import
|
||||
workflowSvc.SetAgentProfileFuncs(
|
||||
buildAgentProfileResolver(repos),
|
||||
buildAgentProfileMatcher(repos),
|
||||
buildAgentProfileMatcher(repos, log),
|
||||
)
|
||||
|
||||
githubSvc := initGitHubService(cfg, dbPool, eventBus, repos.Secrets, log)
|
||||
@@ -1301,26 +1302,93 @@ func buildAgentProfileResolver(repos *Repositories) wfmodels.AgentProfileResolve
|
||||
}
|
||||
}
|
||||
|
||||
// buildAgentProfileMatcher creates a matcher that finds profiles by agent name, model, and mode for import.
|
||||
func buildAgentProfileMatcher(repos *Repositories) wfmodels.AgentProfileMatcher {
|
||||
return func(agentName, model, mode string) string {
|
||||
agents, err := repos.AgentSettings.ListAgents(context.Background())
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
for _, agent := range agents {
|
||||
profiles, pErr := repos.AgentSettings.ListAgentProfiles(context.Background(), agent.ID)
|
||||
if pErr != nil {
|
||||
continue
|
||||
}
|
||||
for _, p := range profiles {
|
||||
if p.AgentDisplayName == agentName && p.Model == model && p.Mode == mode {
|
||||
return p.ID
|
||||
}
|
||||
}
|
||||
// agentProfileStillMatches reports whether the profile with id still has the
|
||||
// exact (agent_display_name, model, mode) triple, ignoring Enabled and
|
||||
// WorkspaceID - those only gate candidate *selection*, not a binding that
|
||||
// already exists.
|
||||
func agentProfileStillMatches(repos *Repositories, id, agentName, model, mode string) bool {
|
||||
p, err := repos.AgentSettings.GetAgentProfile(context.Background(), id)
|
||||
if err != nil || p == nil {
|
||||
return false
|
||||
}
|
||||
return p.AgentDisplayName == agentName && p.Model == model && p.Mode == mode
|
||||
}
|
||||
|
||||
// buildAgentProfileMatcher creates a matcher that finds profiles by agent
|
||||
// name, model, and mode for import.
|
||||
//
|
||||
// The (agent_display_name, model, mode) triple is not unique: duplicating a
|
||||
// profile through the UI produces a byte-identical triple for the copy.
|
||||
// Candidates are filtered to enabled, global (non-workspace-scoped) profiles,
|
||||
// and ties are broken by oldest CreatedAt (then ID for a total order). Oldest
|
||||
// wins so that duplicating a profile can never steal an existing synced
|
||||
// workflow step's binding - a copy is always newer than its source.
|
||||
//
|
||||
// currentID, when non-empty, is checked first: if that profile still has the
|
||||
// exact descriptor, it's kept as-is without re-running candidate selection -
|
||||
// even when it's disabled or workspace-scoped. Reconciliation must not treat
|
||||
// "profile got disabled" as "profile needs a new binding" (profile-disable.md
|
||||
// promises existing bindings survive disabling); candidate selection only
|
||||
// applies when picking a profile for new work.
|
||||
func buildAgentProfileMatcher(repos *Repositories, log *logger.Logger) wfmodels.AgentProfileMatcher {
|
||||
return func(agentName, model, mode, currentID string) string {
|
||||
if currentID != "" && agentProfileStillMatches(repos, currentID, agentName, model, mode) {
|
||||
return currentID
|
||||
}
|
||||
return selectAgentProfileCandidate(repos, log, agentName, model, mode)
|
||||
}
|
||||
}
|
||||
|
||||
// selectAgentProfileCandidate scans enabled, global profiles for an exact
|
||||
// (agent_display_name, model, mode) match and returns the oldest one (see
|
||||
// buildAgentProfileMatcher), logging when the triple was ambiguous.
|
||||
func selectAgentProfileCandidate(repos *Repositories, log *logger.Logger, agentName, model, mode string) string {
|
||||
agents, err := repos.AgentSettings.ListAgents(context.Background())
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
var best *agentsettingsmodels.AgentProfile
|
||||
matches := 0
|
||||
for _, agent := range agents {
|
||||
profiles, pErr := repos.AgentSettings.ListAgentProfiles(context.Background(), agent.ID)
|
||||
if pErr != nil {
|
||||
continue
|
||||
}
|
||||
for _, p := range profiles {
|
||||
if p.AgentDisplayName != agentName || p.Model != model || p.Mode != mode {
|
||||
continue
|
||||
}
|
||||
if !p.Enabled || p.WorkspaceID != "" {
|
||||
continue
|
||||
}
|
||||
matches++
|
||||
if best == nil || isOlderAgentProfileMatch(p, best) {
|
||||
best = p
|
||||
}
|
||||
}
|
||||
}
|
||||
if best == nil {
|
||||
return ""
|
||||
}
|
||||
if matches > 1 {
|
||||
log.Debug("agent profile matcher: multiple candidates for workflow step sync",
|
||||
zap.String("agent_display_name", agentName),
|
||||
zap.String("model", model),
|
||||
zap.String("mode", mode),
|
||||
zap.Int("candidates", matches),
|
||||
zap.String("selected_profile_id", best.ID))
|
||||
}
|
||||
return best.ID
|
||||
}
|
||||
|
||||
// isOlderAgentProfileMatch reports whether candidate should replace current
|
||||
// as the matcher's selection: earlier CreatedAt wins, ties broken by ID so
|
||||
// the result is stable across repeated calls.
|
||||
func isOlderAgentProfileMatch(candidate, current *agentsettingsmodels.AgentProfile) bool {
|
||||
if !candidate.CreatedAt.Equal(current.CreatedAt) {
|
||||
return candidate.CreatedAt.Before(current.CreatedAt)
|
||||
}
|
||||
return candidate.ID < current.ID
|
||||
}
|
||||
|
||||
// codeHostBranchListerAdapter routes first-party GitHub repositories directly
|
||||
|
||||
@@ -29,8 +29,13 @@ type AgentProfilePortable struct {
|
||||
// AgentProfileResolver resolves an agent profile ID to its portable representation.
|
||||
type AgentProfileResolver func(profileID string) *AgentProfilePortable
|
||||
|
||||
// AgentProfileMatcher finds a matching agent profile ID by agent name, model, and mode.
|
||||
type AgentProfileMatcher func(agentName, model, mode string) string
|
||||
// AgentProfileMatcher finds a matching agent profile ID by agent name, model,
|
||||
// and mode. currentID, when non-empty, is the profile already bound (e.g.
|
||||
// during workflow-sync reconciliation); implementations should keep it when
|
||||
// it still matches the descriptor even if the profile was since disabled -
|
||||
// disabling only hides a profile from new selection, it doesn't touch
|
||||
// existing bindings (docs/specs/agents/profile-disable.md).
|
||||
type AgentProfileMatcher func(agentName, model, mode, currentID string) string
|
||||
|
||||
// WorkflowPortable is a workflow without instance-specific fields (IDs, timestamps).
|
||||
type WorkflowPortable struct {
|
||||
@@ -322,7 +327,7 @@ func ConvertReviewProfileToID(events StepEvents, matchProfile AgentProfileMatche
|
||||
agentName, _ := descriptor["agent_name"].(string)
|
||||
model, _ := descriptor["model"].(string)
|
||||
mode, _ := descriptor["mode"].(string)
|
||||
profileID := matchProfile(agentName, model, mode)
|
||||
profileID := matchProfile(agentName, model, mode, "")
|
||||
if profileID == "" {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
@@ -81,7 +81,7 @@ func TestConvertReviewProfileToPortable_NilResolverAndEmptyConfig(t *testing.T)
|
||||
}
|
||||
|
||||
func TestConvertReviewProfileToID_MatchesLocalProfile(t *testing.T) {
|
||||
match := func(agentName, model, mode string) string {
|
||||
match := func(agentName, model, mode, _ string) string {
|
||||
if agentName == "codex" && model == "gpt-5.3" && mode == "review" {
|
||||
return "profile-imported"
|
||||
}
|
||||
@@ -106,7 +106,7 @@ func TestConvertReviewProfileToID_MatchesLocalProfile(t *testing.T) {
|
||||
func TestConvertReviewProfileToID_DropsUnmatchedDescriptor(t *testing.T) {
|
||||
events := ConvertReviewProfileToID(reviewStepEvents(map[string]any{
|
||||
ReviewAgentProfilePortableKey: map[string]any{"agent_name": "unknown"},
|
||||
}), func(string, string, string) string { return "" })
|
||||
}), func(string, string, string, string) string { return "" })
|
||||
|
||||
config := reviewActionConfig(t, events)
|
||||
if _, present := config[ReviewAgentProfileConfigKey]; present {
|
||||
@@ -121,7 +121,7 @@ func TestReviewProfileRoundTripAcrossWorkspaces(t *testing.T) {
|
||||
resolve := func(string) *AgentProfilePortable {
|
||||
return &AgentProfilePortable{AgentName: "claude", Model: "haiku"}
|
||||
}
|
||||
match := func(agentName, model, _ string) string {
|
||||
match := func(agentName, model, _, _ string) string {
|
||||
if agentName == "claude" && model == "haiku" {
|
||||
return "profile-in-target-workspace"
|
||||
}
|
||||
|
||||
@@ -706,7 +706,7 @@ func (s *Service) importSingleWorkflow(ctx context.Context, workspaceID string,
|
||||
// any step. This keeps imports atomic with respect to validation failures.
|
||||
steps := make([]*models.WorkflowStep, 0, len(pw.Steps))
|
||||
for _, sp := range pw.Steps {
|
||||
step := s.stepFromPortable("pending-workflow", sp, posToID)
|
||||
step := s.stepFromPortable("pending-workflow", sp, posToID, "")
|
||||
if err := models.ValidateWorkflowStep(step); err != nil {
|
||||
return nil, fmt.Errorf("validate step %q: %w", sp.Name, err)
|
||||
}
|
||||
@@ -721,7 +721,7 @@ func (s *Service) importSingleWorkflow(ctx context.Context, workspaceID string,
|
||||
needsUpdate := false
|
||||
// Match workflow-level agent profile if present.
|
||||
if pw.AgentProfile != nil && s.matchProfile != nil {
|
||||
if profileID := s.matchProfile(pw.AgentProfile.AgentName, pw.AgentProfile.Model, pw.AgentProfile.Mode); profileID != "" {
|
||||
if profileID := s.matchProfile(pw.AgentProfile.AgentName, pw.AgentProfile.Model, pw.AgentProfile.Mode, ""); profileID != "" {
|
||||
wf.AgentProfileID = profileID
|
||||
needsUpdate = true
|
||||
}
|
||||
@@ -747,8 +747,19 @@ func (s *Service) importSingleWorkflow(ctx context.Context, workspaceID string,
|
||||
|
||||
// stepFromPortable builds a WorkflowStep from its portable form, remapping
|
||||
// position-based references to the step IDs in posToID and matching the
|
||||
// step-level agent profile when a matcher is wired.
|
||||
func (s *Service) stepFromPortable(workflowID string, sp models.StepPortable, posToID map[int]string) *models.WorkflowStep {
|
||||
// step-level agent profile when a matcher is wired. existingProfileID is the
|
||||
// profile already bound to this step (by name) before sync, or "" for a step
|
||||
// with no prior binding (including at import time) - it lets the matcher
|
||||
// preserve a disabled-but-still-matching binding instead of treating
|
||||
// disablement as a rebind.
|
||||
func (s *Service) stepFromPortable(workflowID string, sp models.StepPortable, posToID map[int]string, existingProfileID string) *models.WorkflowStep {
|
||||
return s.stepFromPortableWithMatcher(workflowID, sp, posToID, s.matchProfile, existingProfileID)
|
||||
}
|
||||
|
||||
// stepFromPortableWithMatcher lets workflow sync use a matcher that preserves
|
||||
// profile IDs embedded in existing review actions while imports use the normal
|
||||
// matcher directly.
|
||||
func (s *Service) stepFromPortableWithMatcher(workflowID string, sp models.StepPortable, posToID map[int]string, matchProfile models.AgentProfileMatcher, existingProfileID string) *models.WorkflowStep {
|
||||
step := &models.WorkflowStep{
|
||||
ID: posToID[sp.Position],
|
||||
WorkflowID: workflowID,
|
||||
@@ -756,7 +767,7 @@ func (s *Service) stepFromPortable(workflowID string, sp models.StepPortable, po
|
||||
Position: sp.Position,
|
||||
Color: sp.Color,
|
||||
Prompt: sp.Prompt,
|
||||
Events: models.ConvertReviewProfileToID(models.ConvertPositionToStepID(sp.Events, posToID), s.matchProfile),
|
||||
Events: models.ConvertReviewProfileToID(models.ConvertPositionToStepID(sp.Events, posToID), matchProfile),
|
||||
IsStartStep: sp.IsStartStep,
|
||||
ShowInCommandPanel: sp.ShowInCommandPanel,
|
||||
AllowManualMove: sp.AllowManualMove,
|
||||
@@ -766,8 +777,8 @@ func (s *Service) stepFromPortable(workflowID string, sp models.StepPortable, po
|
||||
WIPLimit: sp.WIPLimit,
|
||||
PullFromStepID: sp.PullFromStepID(posToID),
|
||||
}
|
||||
if sp.AgentProfile != nil && s.matchProfile != nil {
|
||||
step.AgentProfileID = s.matchProfile(sp.AgentProfile.AgentName, sp.AgentProfile.Model, sp.AgentProfile.Mode)
|
||||
if sp.AgentProfile != nil && matchProfile != nil {
|
||||
step.AgentProfileID = matchProfile(sp.AgentProfile.AgentName, sp.AgentProfile.Model, sp.AgentProfile.Mode, existingProfileID)
|
||||
}
|
||||
return step
|
||||
}
|
||||
|
||||
@@ -19,7 +19,10 @@ import (
|
||||
"github.com/kandev/kandev/internal/workflow/repository"
|
||||
)
|
||||
|
||||
func setupTestService(t *testing.T) (*Service, *sqlx.DB) {
|
||||
// setupTestService builds a Service against an in-memory SQLite DB. Callers
|
||||
// needing to observe log output (e.g. via a zaptest observer core) may pass a
|
||||
// logOverride; otherwise a quiet default logger is used.
|
||||
func setupTestService(t *testing.T, logOverride ...*logger.Logger) (*Service, *sqlx.DB) {
|
||||
rawDB, err := sql.Open("sqlite3", ":memory:")
|
||||
require.NoError(t, err)
|
||||
// Pin the pool to one connection: every connection to an in-memory SQLite
|
||||
@@ -41,8 +44,14 @@ func setupTestService(t *testing.T) (*Service, *sqlx.DB) {
|
||||
repo, err := repository.NewWithDB(sqlxDB, sqlxDB, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
|
||||
svc := NewService(repo, log)
|
||||
log := logOverride
|
||||
var svcLogger *logger.Logger
|
||||
if len(log) > 0 {
|
||||
svcLogger = log[0]
|
||||
} else {
|
||||
svcLogger, _ = logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
|
||||
}
|
||||
svc := NewService(repo, svcLogger)
|
||||
t.Cleanup(func() { _ = svc.Close() })
|
||||
return svc, sqlxDB
|
||||
}
|
||||
@@ -57,6 +66,10 @@ func insertWorkflow(t *testing.T, db *sqlx.DB, id, name string) {
|
||||
type mockWorkflowProvider struct {
|
||||
workflows []*taskmodels.Workflow
|
||||
getWorkflowCalls int
|
||||
// forceUpdateWorkflowErr, when set, makes UpdateWorkflow fail without
|
||||
// mutating state - used to test that a rebind Warn never fires for a
|
||||
// change that was not actually persisted.
|
||||
forceUpdateWorkflowErr error
|
||||
}
|
||||
|
||||
func (m *mockWorkflowProvider) ListWorkflows(_ context.Context, workspaceID string, includeHidden bool) ([]*taskmodels.Workflow, error) {
|
||||
@@ -98,6 +111,9 @@ func (m *mockWorkflowProvider) CreateWorkflow(_ context.Context, workspaceID, na
|
||||
}
|
||||
|
||||
func (m *mockWorkflowProvider) UpdateWorkflow(_ context.Context, workflow *taskmodels.Workflow) error {
|
||||
if m.forceUpdateWorkflowErr != nil {
|
||||
return m.forceUpdateWorkflowErr
|
||||
}
|
||||
for i, wf := range m.workflows {
|
||||
if wf.ID == workflow.ID {
|
||||
m.workflows[i] = workflow
|
||||
@@ -919,7 +935,7 @@ func TestImportWorkflows(t *testing.T) {
|
||||
svc, _, _ := setupTestServiceWithProvider(t)
|
||||
ctx := context.Background()
|
||||
|
||||
matcher := func(agentName, model, mode string) string {
|
||||
matcher := func(agentName, model, mode, _ string) string {
|
||||
if agentName == "Claude Code" && model == "opus" {
|
||||
return "matched-prof-1"
|
||||
}
|
||||
|
||||
@@ -160,7 +160,7 @@ func (s *Service) updateSyncedWorkflow(ctx context.Context, wf *taskmodels.Workf
|
||||
// persisting any step. A malformed later step must leave the synced workflow
|
||||
// untouched so the next sync can retry after the source is fixed.
|
||||
for _, sp := range pw.Steps {
|
||||
step := s.stepFromPortable(wf.ID, sp, posToID)
|
||||
step := s.stepFromPortableForSync(wf.ID, sp, posToID, existingByName[sp.Name])
|
||||
if err := models.ValidateWorkflowStep(step); err != nil {
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("workflow %q: invalid step %q: %v", wf.Name, sp.Name, err))
|
||||
return
|
||||
@@ -173,13 +173,21 @@ func (s *Service) updateSyncedWorkflow(ctx context.Context, wf *taskmodels.Workf
|
||||
return
|
||||
}
|
||||
for _, sp := range pw.Steps {
|
||||
step := s.stepFromPortable(wf.ID, sp, posToID)
|
||||
if existing, ok := existingByName[sp.Name]; ok {
|
||||
existing := existingByName[sp.Name]
|
||||
step := s.stepFromPortableForSync(wf.ID, sp, posToID, existing)
|
||||
var rebinding bool
|
||||
var oldProfileID, existingID string
|
||||
var reviewRebindings []profileRebinding
|
||||
if existing != nil {
|
||||
step.CreatedAt = existing.CreatedAt
|
||||
step.StageType = existing.StageType // not carried by the portable format
|
||||
if stepMatchesDefinition(existing, step) {
|
||||
continue
|
||||
}
|
||||
rebinding = existing.AgentProfileID != "" && existing.AgentProfileID != step.AgentProfileID
|
||||
oldProfileID = existing.AgentProfileID
|
||||
existingID = existing.ID
|
||||
reviewRebindings = findReviewProfileRebindings(existing.Events, step.Events)
|
||||
err = s.repo.UpdateStep(ctx, step)
|
||||
} else {
|
||||
err = s.repo.CreateStep(ctx, step)
|
||||
@@ -188,6 +196,7 @@ func (s *Service) updateSyncedWorkflow(ctx context.Context, wf *taskmodels.Workf
|
||||
result.Warnings = append(result.Warnings, fmt.Sprintf("workflow %q: failed to apply step %q: %v", wf.Name, sp.Name, err))
|
||||
return
|
||||
}
|
||||
s.logStepProfileRebindings(wf, sp.Name, existingID, rebinding, oldProfileID, step.AgentProfileID, reviewRebindings)
|
||||
changed = true
|
||||
}
|
||||
for _, st := range toDelete {
|
||||
@@ -202,6 +211,142 @@ func (s *Service) updateSyncedWorkflow(ctx context.Context, wf *taskmodels.Workf
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) logStepProfileRebindings(wf *taskmodels.Workflow, stepName, stepID string, rebinding bool, oldProfileID, newProfileID string, reviewRebindings []profileRebinding) {
|
||||
if rebinding {
|
||||
s.logger.Warn("workflow sync rebinding step's agent profile",
|
||||
zap.String("workflow_id", wf.ID),
|
||||
zap.String("workflow_name", wf.Name),
|
||||
zap.String("step_id", stepID),
|
||||
zap.String("step_name", stepName),
|
||||
zap.String("old_agent_profile_id", oldProfileID),
|
||||
zap.String("new_agent_profile_id", newProfileID))
|
||||
}
|
||||
for _, reviewRebinding := range reviewRebindings {
|
||||
s.logger.Warn("workflow sync rebinding step's review agent profile",
|
||||
zap.String("workflow_id", wf.ID),
|
||||
zap.String("workflow_name", wf.Name),
|
||||
zap.String("step_id", stepID),
|
||||
zap.String("step_name", stepName),
|
||||
zap.String("old_agent_profile_id", reviewRebinding.oldID),
|
||||
zap.String("new_agent_profile_id", reviewRebinding.newID))
|
||||
}
|
||||
}
|
||||
|
||||
type profileRebinding struct {
|
||||
oldID string
|
||||
newID string
|
||||
}
|
||||
|
||||
func (s *Service) stepFromPortableForSync(workflowID string, sp models.StepPortable, posToID map[int]string, existing *models.WorkflowStep) *models.WorkflowStep {
|
||||
existingProfileID := ""
|
||||
if existing != nil {
|
||||
existingProfileID = existing.AgentProfileID
|
||||
}
|
||||
var matchProfile models.AgentProfileMatcher
|
||||
if s.matchProfile != nil {
|
||||
matchProfile = s.reviewProfileMatcherForSync(existing)
|
||||
}
|
||||
return s.stepFromPortableWithMatcher(workflowID, sp, posToID, matchProfile, existingProfileID)
|
||||
}
|
||||
|
||||
func (s *Service) reviewProfileMatcherForSync(existing *models.WorkflowStep) models.AgentProfileMatcher {
|
||||
preserved := make(map[string][]string)
|
||||
if existing != nil && s.resolveProfile != nil {
|
||||
for _, action := range existing.Events.OnEnter {
|
||||
if action.Type != models.OnEnterRunCodeReview || action.Config == nil {
|
||||
continue
|
||||
}
|
||||
profileID, ok := action.Config[models.ReviewAgentProfileConfigKey].(string)
|
||||
if !ok || profileID == "" {
|
||||
continue
|
||||
}
|
||||
if portable := s.resolveProfile(profileID); portable != nil {
|
||||
key := portableProfileKey(portable.AgentName, portable.Model, portable.Mode)
|
||||
preserved[key] = append(preserved[key], profileID)
|
||||
}
|
||||
}
|
||||
}
|
||||
return func(agentName, model, mode, currentID string) string {
|
||||
if currentID != "" {
|
||||
return s.matchProfile(agentName, model, mode, currentID)
|
||||
}
|
||||
key := portableProfileKey(agentName, model, mode)
|
||||
if candidates := preserved[key]; len(candidates) > 0 {
|
||||
profileID := candidates[0]
|
||||
preserved[key] = candidates[1:]
|
||||
return profileID
|
||||
}
|
||||
if s.matchProfile == nil {
|
||||
return ""
|
||||
}
|
||||
return s.matchProfile(agentName, model, mode, "")
|
||||
}
|
||||
}
|
||||
|
||||
func portableProfileKey(agentName, model, mode string) string {
|
||||
return agentName + "\x00" + model + "\x00" + mode
|
||||
}
|
||||
|
||||
func findReviewProfileRebindings(existing, desired models.StepEvents) []profileRebinding {
|
||||
oldIDs := reviewProfileIDs(existing)
|
||||
newIDs := reviewProfileIDs(desired)
|
||||
newCounts := make(map[string]int, len(newIDs))
|
||||
for _, profileID := range newIDs {
|
||||
if profileID != "" {
|
||||
newCounts[profileID]++
|
||||
}
|
||||
}
|
||||
var unmatchedOld []string
|
||||
for _, profileID := range oldIDs {
|
||||
if profileID == "" {
|
||||
continue
|
||||
}
|
||||
if newCounts[profileID] > 0 {
|
||||
newCounts[profileID]--
|
||||
continue
|
||||
}
|
||||
unmatchedOld = append(unmatchedOld, profileID)
|
||||
}
|
||||
oldCounts := make(map[string]int, len(oldIDs))
|
||||
for _, profileID := range oldIDs {
|
||||
if profileID != "" {
|
||||
oldCounts[profileID]++
|
||||
}
|
||||
}
|
||||
var unmatchedNew []string
|
||||
for _, profileID := range newIDs {
|
||||
if profileID == "" {
|
||||
continue
|
||||
}
|
||||
if oldCounts[profileID] > 0 {
|
||||
oldCounts[profileID]--
|
||||
continue
|
||||
}
|
||||
unmatchedNew = append(unmatchedNew, profileID)
|
||||
}
|
||||
var rebindings []profileRebinding
|
||||
for i, oldID := range unmatchedOld {
|
||||
newID := ""
|
||||
if i < len(unmatchedNew) {
|
||||
newID = unmatchedNew[i]
|
||||
}
|
||||
rebindings = append(rebindings, profileRebinding{oldID: oldID, newID: newID})
|
||||
}
|
||||
return rebindings
|
||||
}
|
||||
|
||||
func reviewProfileIDs(events models.StepEvents) []string {
|
||||
var profileIDs []string
|
||||
for _, action := range events.OnEnter {
|
||||
if action.Type != models.OnEnterRunCodeReview || action.Config == nil {
|
||||
continue
|
||||
}
|
||||
profileID, _ := action.Config[models.ReviewAgentProfileConfigKey].(string)
|
||||
profileIDs = append(profileIDs, profileID)
|
||||
}
|
||||
return profileIDs
|
||||
}
|
||||
|
||||
// stepMatchesDefinition reports whether an existing step already equals its
|
||||
// desired definition (timestamps and stage type aside — stage type is copied
|
||||
// from the existing step). Events are compared via their JSON encoding to
|
||||
@@ -268,15 +413,27 @@ func (s *Service) planStepChanges(ctx context.Context, wf *taskmodels.Workflow,
|
||||
func (s *Service) applyWorkflowFields(ctx context.Context, wf *taskmodels.Workflow, pw models.WorkflowPortable) (bool, error) {
|
||||
profileID := ""
|
||||
if pw.AgentProfile != nil && s.matchProfile != nil {
|
||||
profileID = s.matchProfile(pw.AgentProfile.AgentName, pw.AgentProfile.Model, pw.AgentProfile.Mode)
|
||||
profileID = s.matchProfile(pw.AgentProfile.AgentName, pw.AgentProfile.Model, pw.AgentProfile.Mode, wf.AgentProfileID)
|
||||
}
|
||||
if wf.Description == pw.Description && wf.Prompt == pw.Prompt && wf.AgentProfileID == profileID {
|
||||
return false, nil
|
||||
}
|
||||
oldProfileID := wf.AgentProfileID
|
||||
rebinding := oldProfileID != "" && oldProfileID != profileID
|
||||
wf.Description = pw.Description
|
||||
wf.Prompt = pw.Prompt
|
||||
wf.AgentProfileID = profileID
|
||||
return true, s.workflowProvider.UpdateWorkflow(ctx, wf)
|
||||
if err := s.workflowProvider.UpdateWorkflow(ctx, wf); err != nil {
|
||||
return true, err
|
||||
}
|
||||
if rebinding {
|
||||
s.logger.Warn("workflow sync rebinding workflow's agent profile",
|
||||
zap.String("workflow_id", wf.ID),
|
||||
zap.String("workflow_name", wf.Name),
|
||||
zap.String("old_agent_profile_id", oldProfileID),
|
||||
zap.String("new_agent_profile_id", profileID))
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// ReleaseSyncedWorkflows converts every GitHub-sourced workflow in the
|
||||
|
||||
@@ -7,7 +7,11 @@ import (
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
taskmodels "github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/kandev/kandev/internal/workflow/models"
|
||||
)
|
||||
@@ -66,6 +70,17 @@ func setupSyncService(t *testing.T) (*Service, *mockWorkflowProvider, *mockSyncO
|
||||
return svc, provider, ops
|
||||
}
|
||||
|
||||
// setupSyncServiceWithLogger is a variant of setupSyncService for tests that
|
||||
// need to observe log output (e.g. via a zaptest observer core).
|
||||
func setupSyncServiceWithLogger(t *testing.T, log *logger.Logger) (*Service, *mockWorkflowProvider, *mockSyncOps) {
|
||||
svc, _ := setupTestService(t, log)
|
||||
provider := &mockWorkflowProvider{}
|
||||
svc.SetWorkflowProvider(provider)
|
||||
ops := newMockSyncOps(provider)
|
||||
svc.SetSyncWorkflowOps(ops)
|
||||
return svc, provider, ops
|
||||
}
|
||||
|
||||
func portableWorkflow(name string, stepNames ...string) models.WorkflowPortable {
|
||||
steps := make([]models.StepPortable, 0, len(stepNames))
|
||||
for i, sn := range stepNames {
|
||||
@@ -492,3 +507,238 @@ func TestReleaseSyncedWorkflows(t *testing.T) {
|
||||
assert.NoError(t, svc.EnsureWorkflowMutable(ctx, id), "released workflows are editable again")
|
||||
}
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_RebindingStepAgentProfileLogsWarning covers the
|
||||
// visibility half of the profile-rebinding defect: a sync round that changes
|
||||
// an existing step's AgentProfileID must warn (naming the workflow, step,
|
||||
// and both profile IDs) since the workflow is read-only to the user and the
|
||||
// change would otherwise be silent. A second sync that resolves to the same
|
||||
// profile must not add another warning.
|
||||
func TestApplySyncedWorkflows_RebindingStepAgentProfileLogsWarning(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, IsStartStep: true,
|
||||
AgentProfileID: "profile-old",
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "profile-new" })
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"Dev Flow"}, result.Updated)
|
||||
|
||||
entries := logs.FilterMessage("workflow sync rebinding step's agent profile").All()
|
||||
require.Len(t, entries, 1, "rebind must be logged exactly once")
|
||||
fields := entries[0].ContextMap()
|
||||
assert.Equal(t, "step-todo", fields["step_id"])
|
||||
assert.Equal(t, "Todo", fields["step_name"])
|
||||
assert.Equal(t, "profile-old", fields["old_agent_profile_id"])
|
||||
assert.Equal(t, "profile-new", fields["new_agent_profile_id"])
|
||||
|
||||
// A second sync resolving to the same profile must not warn again.
|
||||
result, err = svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, result.Updated, "no drift means nothing is rewritten")
|
||||
assert.Len(t, logs.FilterMessage("workflow sync rebinding step's agent profile").All(), 1,
|
||||
"an unchanged profile must not log another warning")
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_RebindWarningDoesNotFireOnFailedUpdate covers
|
||||
// R1-F1: the rebind Warn must only fire once the change is actually
|
||||
// persisted. It forces the step's UpdateStep write to fail (via PRAGMA
|
||||
// query_only, which makes every write on this connection fail while reads —
|
||||
// like the ListStepsByWorkflow that already ran — keep working) and asserts
|
||||
// no Warn is logged for a rebind that never landed.
|
||||
func TestApplySyncedWorkflows_RebindWarningDoesNotFireOnFailedUpdate(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, db := setupTestService(t, log)
|
||||
provider := &mockWorkflowProvider{}
|
||||
svc.SetWorkflowProvider(provider)
|
||||
ops := newMockSyncOps(provider)
|
||||
svc.SetSyncWorkflowOps(ops)
|
||||
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, IsStartStep: true,
|
||||
AgentProfileID: "profile-old",
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "profile-new" })
|
||||
|
||||
_, err = db.Exec("PRAGMA query_only = ON")
|
||||
t.Cleanup(func() { _, _ = db.Exec("PRAGMA query_only = OFF") })
|
||||
require.NoError(t, err)
|
||||
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, result.Warnings, 1, "the failed step write surfaces a warning")
|
||||
assert.Contains(t, result.Warnings[0], "Todo")
|
||||
assert.Empty(t, result.Updated, "a step update that failed to persist must not be reported as applied")
|
||||
assert.Empty(t, logs.FilterMessage("workflow sync rebinding step's agent profile").All(),
|
||||
"a rebind Warn must never fire for a change that was not actually persisted")
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_RebindingWorkflowAgentProfileLogsWarning covers
|
||||
// R1-F2: the workflow-level AgentProfileID (the fallback steps with no
|
||||
// profile of their own resolve to) is rebound by the same matcher and must
|
||||
// be logged just like a step-level rebind.
|
||||
func TestApplySyncedWorkflows_RebindingWorkflowAgentProfileLogsWarning(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
wf.AgentProfileID = "profile-old"
|
||||
createStep(t, svc, &models.WorkflowStep{ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, IsStartStep: true})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "profile-new" })
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"Dev Flow"}, result.Updated)
|
||||
|
||||
got, err := provider.GetWorkflow(ctx, wf.ID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "profile-new", got.AgentProfileID)
|
||||
|
||||
entries := logs.FilterMessage("workflow sync rebinding workflow's agent profile").All()
|
||||
require.Len(t, entries, 1, "workflow-level rebind must be logged exactly once")
|
||||
fields := entries[0].ContextMap()
|
||||
assert.Equal(t, "wf-1", fields["workflow_id"])
|
||||
assert.Equal(t, "Dev Flow", fields["workflow_name"])
|
||||
assert.Equal(t, "profile-old", fields["old_agent_profile_id"])
|
||||
assert.Equal(t, "profile-new", fields["new_agent_profile_id"])
|
||||
|
||||
// A second sync resolving to the same profile must not warn again.
|
||||
result, err = svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, result.Updated, "no drift means nothing is rewritten")
|
||||
assert.Len(t, logs.FilterMessage("workflow sync rebinding workflow's agent profile").All(), 1,
|
||||
"an unchanged profile must not log another warning")
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_WorkflowRebindWarningDoesNotFireOnFailedUpdate is
|
||||
// the workflow-level counterpart to
|
||||
// TestApplySyncedWorkflows_RebindWarningDoesNotFireOnFailedUpdate (R1-F1): a
|
||||
// rebind Warn must never fire for a workflow-level profile change that failed
|
||||
// to persist.
|
||||
func TestApplySyncedWorkflows_WorkflowRebindWarningDoesNotFireOnFailedUpdate(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
wf.AgentProfileID = "profile-old"
|
||||
createStep(t, svc, &models.WorkflowStep{ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, IsStartStep: true})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "profile-new" })
|
||||
|
||||
provider.forceUpdateWorkflowErr = fmt.Errorf("simulated UpdateWorkflow failure")
|
||||
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, result.Warnings, 1, "the failed workflow write surfaces a warning")
|
||||
assert.Contains(t, result.Warnings[0], "Dev Flow")
|
||||
assert.Empty(t, result.Updated, "a workflow update that failed to persist must not be reported as applied")
|
||||
assert.Empty(t, logs.FilterMessage("workflow sync rebinding workflow's agent profile").All(),
|
||||
"a rebind Warn must never fire for a workflow change that was not actually persisted")
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_StepMatcherReceivesCurrentBinding proves
|
||||
// updateSyncedWorkflow passes the step's existing AgentProfileID to the
|
||||
// matcher as currentID. A real matcher (buildAgentProfileMatcher) uses that
|
||||
// to preserve a binding that was disabled after the fact instead of treating
|
||||
// disablement as a rebind (docs/specs/agents/profile-disable.md). This test
|
||||
// stubs that same contract: the matcher echoes currentID back when given
|
||||
// one, standing in for "the bound profile still matches, keep it".
|
||||
func TestApplySyncedWorkflows_StepMatcherReceivesCurrentBinding(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
wf.Description = "synced Dev Flow" // must already match pw's description below, or the workflow itself counts as changed
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, Color: "#aabbcc", IsStartStep: true,
|
||||
AgentProfileID: "profile-disabled",
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
|
||||
svc.SetAgentProfileFuncs(nil, func(_, _, _, currentID string) string {
|
||||
if currentID != "" {
|
||||
return currentID
|
||||
}
|
||||
return "profile-new"
|
||||
})
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, result.Updated, "a preserved binding must not be reported as a change")
|
||||
assert.Empty(t, logs.FilterMessage("workflow sync rebinding step's agent profile").All(),
|
||||
"a preserved binding must not log a rebind")
|
||||
}
|
||||
|
||||
// TestApplySyncedWorkflows_StepProfileUnbindWarnsWithEmptyNewID covers a
|
||||
// genuine unbind: the matcher finds no candidate at all (not the
|
||||
// still-matches-but-disabled case above), so the step's AgentProfileID
|
||||
// resolves to "". The rebind Warn must still fire and its
|
||||
// new_agent_profile_id field must be observably empty, so a reader of the
|
||||
// log can tell the step lost its profile rather than gained a different one.
|
||||
func TestApplySyncedWorkflows_StepProfileUnbindWarnsWithEmptyNewID(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, IsStartStep: true,
|
||||
AgentProfileID: "profile-old",
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].AgentProfile = &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
files := []SyncFileExport{{Path: "flows/dev.yml", Export: exportOf(pw)}}
|
||||
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "" })
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", files)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"Dev Flow"}, result.Updated)
|
||||
|
||||
entries := logs.FilterMessage("workflow sync rebinding step's agent profile").All()
|
||||
require.Len(t, entries, 1, "an unbind must still be logged")
|
||||
fields := entries[0].ContextMap()
|
||||
assert.Equal(t, "profile-old", fields["old_agent_profile_id"])
|
||||
assert.Equal(t, "", fields["new_agent_profile_id"])
|
||||
}
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/workflow/models"
|
||||
)
|
||||
|
||||
func TestApplySyncedWorkflows_PreservesCurrentReviewAgentProfileBinding(t *testing.T) {
|
||||
svc, provider, _ := setupSyncService(t)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
wf.Description = "synced Dev Flow"
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, Color: "#aabbcc", IsStartStep: true,
|
||||
Events: models.StepEvents{OnEnter: []models.OnEnterAction{{
|
||||
Type: models.OnEnterRunCodeReview,
|
||||
Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-current"},
|
||||
}}},
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].Events = models.StepEvents{OnEnter: []models.OnEnterAction{{
|
||||
Type: models.OnEnterRunCodeReview,
|
||||
Config: map[string]interface{}{models.ReviewAgentProfilePortableKey: map[string]interface{}{
|
||||
"agent_name": "Claude", "model": "opus[1m]", "mode": "auto",
|
||||
}},
|
||||
}}}
|
||||
svc.SetAgentProfileFuncs(func(id string) *models.AgentProfilePortable {
|
||||
if id == "profile-current" {
|
||||
return &models.AgentProfilePortable{AgentName: "Claude", Model: "opus[1m]", Mode: "auto"}
|
||||
}
|
||||
return nil
|
||||
}, func(string, string, string, string) string { return "profile-oldest" })
|
||||
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", []SyncFileExport{{
|
||||
Path: "flows/dev.yml", Export: exportOf(pw),
|
||||
}})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, result.Updated, "an existing exact review binding must win over a global candidate")
|
||||
|
||||
steps, err := svc.ListStepsByWorkflow(ctx, wf.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, steps, 1)
|
||||
require.Len(t, steps[0].Events.OnEnter, 1)
|
||||
assert.Equal(t, "profile-current", steps[0].Events.OnEnter[0].Config[models.ReviewAgentProfileConfigKey])
|
||||
}
|
||||
|
||||
func TestApplySyncedWorkflows_LogsReviewAgentProfileRebinding(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, provider, _ := setupSyncServiceWithLogger(t, log)
|
||||
ctx := context.Background()
|
||||
wf := addSyncedWorkflow(provider, "wf-1", "ws-1", "Dev Flow", "flows/dev.yml")
|
||||
wf.Description = "synced Dev Flow"
|
||||
createStep(t, svc, &models.WorkflowStep{
|
||||
ID: "step-todo", WorkflowID: wf.ID, Name: "Todo", Position: 0, Color: "#aabbcc", IsStartStep: true,
|
||||
Events: models.StepEvents{OnEnter: []models.OnEnterAction{{
|
||||
Type: models.OnEnterRunCodeReview,
|
||||
Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-old"},
|
||||
}}},
|
||||
})
|
||||
|
||||
pw := portableWorkflow("Dev Flow", "Todo")
|
||||
pw.Steps[0].Events = models.StepEvents{OnEnter: []models.OnEnterAction{{
|
||||
Type: models.OnEnterRunCodeReview,
|
||||
Config: map[string]interface{}{models.ReviewAgentProfilePortableKey: map[string]interface{}{
|
||||
"agent_name": "Claude", "model": "opus[1m]", "mode": "auto",
|
||||
}},
|
||||
}}}
|
||||
svc.SetAgentProfileFuncs(nil, func(string, string, string, string) string { return "profile-new" })
|
||||
|
||||
result, err := svc.ApplySyncedWorkflows(ctx, "ws-1", []SyncFileExport{{
|
||||
Path: "flows/dev.yml", Export: exportOf(pw),
|
||||
}})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"Dev Flow"}, result.Updated)
|
||||
|
||||
entries := logs.FilterMessage("workflow sync rebinding step's review agent profile").All()
|
||||
require.Len(t, entries, 1)
|
||||
assert.Equal(t, "profile-old", entries[0].ContextMap()["old_agent_profile_id"])
|
||||
assert.Equal(t, "profile-new", entries[0].ContextMap()["new_agent_profile_id"])
|
||||
}
|
||||
|
||||
func TestFindReviewProfileRebindingsIgnoresActionOrder(t *testing.T) {
|
||||
existing := models.StepEvents{OnEnter: []models.OnEnterAction{
|
||||
{Type: models.OnEnterRunCodeReview, Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-a"}},
|
||||
{Type: models.OnEnterRunCodeReview, Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-b"}},
|
||||
}}
|
||||
desired := models.StepEvents{OnEnter: []models.OnEnterAction{
|
||||
{Type: models.OnEnterRunCodeReview, Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-b"}},
|
||||
{Type: models.OnEnterRunCodeReview, Config: map[string]interface{}{models.ReviewAgentProfileConfigKey: "profile-a"}},
|
||||
}}
|
||||
|
||||
assert.Empty(t, findReviewProfileRebindings(existing, desired))
|
||||
}
|
||||
Reference in New Issue
Block a user