fix: honor same-profile workflow session policy (#3605)
This commit is contained in:
@@ -680,6 +680,7 @@ func startAgentInfrastructure(
|
||||
log.Error("Failed to initialize orchestrator", zap.Error(err))
|
||||
return false
|
||||
}
|
||||
services.Task.SetWorkflowMovePreflight(orchestratorSvc)
|
||||
orchestratorSvc.SetAgentctlBinaryPath(agentctlBinaryPath)
|
||||
orchestratorSvc.SetRouteActionHandler(dynamicRouteActionHandler(
|
||||
repos.Task,
|
||||
|
||||
@@ -2921,6 +2921,17 @@ func (s *Service) resolveStepProfileSessionStartPolicy(step *wfmodels.WorkflowSt
|
||||
return models.NormalizeWorkflowProfileSessionStartPolicy(string(step.ProfileSessionStartPolicy))
|
||||
}
|
||||
|
||||
// shouldKeepCurrentWorkflowStepSession reports whether profile-only routing
|
||||
// can keep the current session without applying replacement policy.
|
||||
func shouldKeepCurrentWorkflowStepSession(
|
||||
effectiveProfile string,
|
||||
currentProfile string,
|
||||
startPolicy models.WorkflowProfileSessionStartPolicy,
|
||||
) bool {
|
||||
return effectiveProfile == "" ||
|
||||
(effectiveProfile == currentProfile && startPolicy == models.WorkflowProfileSessionStartPolicyReuse)
|
||||
}
|
||||
|
||||
// resolveStepProfileSessionEndPolicy returns the source step's session end
|
||||
// behavior. Invalid or absent values use the conversation-preserving park default.
|
||||
func (s *Service) resolveStepProfileSessionEndPolicy(step *wfmodels.WorkflowStep) models.WorkflowProfileSessionEndPolicy {
|
||||
@@ -3737,25 +3748,10 @@ func (s *Service) prepareWorkflowStepSession(
|
||||
return s.prepareExplicitWorkflowSession(ctx, taskID, session, step, sourceStep, entryIDs...)
|
||||
}
|
||||
effectiveProfile := s.resolveStepAgentProfile(ctx, step)
|
||||
if effectiveProfile == "" || effectiveProfile == session.AgentProfileID {
|
||||
s.tagSessionAsWorkflowSwitched(ctx, session.ID)
|
||||
if !session.IsPrimary {
|
||||
if err := s.SetPrimarySession(ctx, session.ID); err != nil {
|
||||
s.logger.Warn("failed to preserve session as primary for workflow step",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", session.ID),
|
||||
zap.String("step_id", step.ID),
|
||||
zap.Error(err))
|
||||
} else {
|
||||
session.IsPrimary = true
|
||||
}
|
||||
}
|
||||
if err := s.recordWorkflowSourceBinding(ctx, taskID, step, session, entryIDs...); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return session, false, nil
|
||||
}
|
||||
startPolicy := s.resolveStepProfileSessionStartPolicy(step)
|
||||
if shouldKeepCurrentWorkflowStepSession(effectiveProfile, session.AgentProfileID, startPolicy) {
|
||||
return s.keepCurrentWorkflowStepSession(ctx, taskID, session, step, entryIDs...)
|
||||
}
|
||||
if sourceStep == nil {
|
||||
return nil, false, fmt.Errorf("workflow profile switch source step is unavailable")
|
||||
}
|
||||
@@ -3770,6 +3766,31 @@ func (s *Service) prepareWorkflowStepSession(
|
||||
return newSession, true, nil
|
||||
}
|
||||
|
||||
func (s *Service) keepCurrentWorkflowStepSession(
|
||||
ctx context.Context,
|
||||
taskID string,
|
||||
session *models.TaskSession,
|
||||
step *wfmodels.WorkflowStep,
|
||||
entryIDs ...int64,
|
||||
) (*models.TaskSession, bool, error) {
|
||||
s.tagSessionAsWorkflowSwitched(ctx, session.ID)
|
||||
if !session.IsPrimary {
|
||||
if err := s.SetPrimarySession(ctx, session.ID); err != nil {
|
||||
s.logger.Warn("failed to preserve session as primary for workflow step",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", session.ID),
|
||||
zap.String("step_id", step.ID),
|
||||
zap.Error(err))
|
||||
} else {
|
||||
session.IsPrimary = true
|
||||
}
|
||||
}
|
||||
if err := s.recordWorkflowSourceBinding(ctx, taskID, step, session, entryIDs...); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return session, false, nil
|
||||
}
|
||||
|
||||
func (s *Service) preflightWorkflowStepCredentials(
|
||||
ctx context.Context,
|
||||
taskID string,
|
||||
@@ -3801,10 +3822,10 @@ func (s *Service) preflightWorkflowStepCredentials(
|
||||
)
|
||||
}
|
||||
effectiveProfile := s.resolveStepAgentProfile(ctx, targetStep)
|
||||
if effectiveProfile == "" || effectiveProfile == currentSession.AgentProfileID {
|
||||
startPolicy := s.resolveStepProfileSessionStartPolicy(targetStep)
|
||||
if shouldKeepCurrentWorkflowStepSession(effectiveProfile, currentSession.AgentProfileID, startPolicy) {
|
||||
return nil
|
||||
}
|
||||
startPolicy := s.resolveStepProfileSessionStartPolicy(targetStep)
|
||||
targetSession := currentSession
|
||||
if startPolicy == models.WorkflowProfileSessionStartPolicyReuse {
|
||||
existing, err := s.findReusableSessionForProfile(ctx, taskID, effectiveProfile, currentSession.ID)
|
||||
@@ -3827,6 +3848,17 @@ func (s *Service) preflightWorkflowStepCredentials(
|
||||
)
|
||||
}
|
||||
|
||||
// PreflightWorkflowStepMove exposes the destination lifecycle preflight to
|
||||
// the task service, which must run it before committing a manual move.
|
||||
func (s *Service) PreflightWorkflowStepMove(
|
||||
ctx context.Context,
|
||||
taskID string,
|
||||
currentSession *models.TaskSession,
|
||||
targetStep *wfmodels.WorkflowStep,
|
||||
) error {
|
||||
return s.preflightWorkflowStepCredentials(ctx, taskID, currentSession, targetStep)
|
||||
}
|
||||
|
||||
// maybySwitchSessionForProfile preserves the legacy processOnEnter failure
|
||||
// handling while sharing the error-returning step-entry preflight with direct
|
||||
// workflow-engine dispatch.
|
||||
|
||||
+216
@@ -0,0 +1,216 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
wfmodels "github.com/kandev/kandev/internal/workflow/models"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPrepareWorkflowStepSession_SameProfileNew(t *testing.T) {
|
||||
for _, endPolicy := range []models.WorkflowProfileSessionEndPolicy{
|
||||
models.WorkflowProfileSessionEndPolicyComplete,
|
||||
models.WorkflowProfileSessionEndPolicyPark,
|
||||
} {
|
||||
t.Run(string(endPolicy), func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newProfileSwitchFixture(t, models.WorkflowProfileSessionStartPolicyNew, endPolicy)
|
||||
extra := &models.TaskSession{
|
||||
ID: "session-a-existing", TaskID: "t1", AgentProfileID: "profile-a",
|
||||
ExecutorID: "exec-local", ExecutorProfileID: "ep1", TaskEnvironmentID: "env-1",
|
||||
State: models.TaskSessionStateWaitingForInput,
|
||||
StartedAt: time.Now().UTC().Add(-time.Minute), UpdatedAt: time.Now().UTC().Add(-time.Minute),
|
||||
}
|
||||
require.NoError(t, fixture.repo.CreateTaskSession(ctx, extra))
|
||||
|
||||
target := &wfmodels.WorkflowStep{
|
||||
ID: "step-b", WorkflowID: "wf1", Position: 1, AgentProfileID: "profile-a",
|
||||
ProfileSessionStartPolicy: models.WorkflowProfileSessionStartPolicyNew,
|
||||
}
|
||||
source := &wfmodels.WorkflowStep{
|
||||
ID: "step-a", WorkflowID: "wf1", Position: 0, AgentProfileID: "profile-a",
|
||||
ProfileSessionEndPolicy: endPolicy,
|
||||
}
|
||||
|
||||
selected, switched, err := fixture.svc.prepareWorkflowStepSession(
|
||||
ctx, "t1", fixture.current, target, source,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.True(t, switched)
|
||||
require.NotNil(t, selected)
|
||||
require.NotEqual(t, fixture.current.ID, selected.ID)
|
||||
require.NotEqual(t, extra.ID, selected.ID)
|
||||
require.Equal(t, "profile-a", selected.AgentProfileID)
|
||||
|
||||
current, err := fixture.repo.GetTaskSession(ctx, fixture.current.ID)
|
||||
require.NoError(t, err)
|
||||
switch endPolicy {
|
||||
case models.WorkflowProfileSessionEndPolicyComplete:
|
||||
require.Equal(t, models.TaskSessionStateCompleted, current.State)
|
||||
require.NotNil(t, current.CompletedAt)
|
||||
case models.WorkflowProfileSessionEndPolicyPark:
|
||||
require.Equal(t, models.TaskSessionStateWaitingForInput, current.State)
|
||||
require.Nil(t, current.CompletedAt)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareWorkflowStepSession_SameProfileReusePreservesOverrides(t *testing.T) {
|
||||
for _, startPolicy := range []models.WorkflowProfileSessionStartPolicy{
|
||||
models.WorkflowProfileSessionStartPolicyReuse,
|
||||
"",
|
||||
} {
|
||||
t.Run(string(startPolicy), func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newProfileSwitchFixture(t, models.WorkflowProfileSessionStartPolicyReuse, models.WorkflowProfileSessionEndPolicyPark)
|
||||
overrides := models.SessionRuntimeConfig{
|
||||
Model: "astra",
|
||||
ConfigOptions: map[string]string{"reasoning_effort": "medium"},
|
||||
}
|
||||
require.NoError(t, fixture.repo.SetSessionMetadataKey(
|
||||
ctx, fixture.current.ID, models.SessionMetaKeyRuntimeConfigOverrides, overrides,
|
||||
))
|
||||
|
||||
target := &wfmodels.WorkflowStep{
|
||||
ID: "step-b", WorkflowID: "wf1", Position: 1, AgentProfileID: "profile-a",
|
||||
ProfileSessionStartPolicy: startPolicy,
|
||||
}
|
||||
source := &wfmodels.WorkflowStep{
|
||||
ID: "step-a", WorkflowID: "wf1", Position: 0, AgentProfileID: "profile-a",
|
||||
ProfileSessionEndPolicy: models.WorkflowProfileSessionEndPolicyPark,
|
||||
}
|
||||
|
||||
selected, switched, err := fixture.svc.prepareWorkflowStepSession(
|
||||
ctx, "t1", fixture.current, target, source,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.False(t, switched)
|
||||
require.Equal(t, fixture.current.ID, selected.ID)
|
||||
|
||||
updated, err := fixture.repo.GetTaskSession(ctx, fixture.current.ID)
|
||||
require.NoError(t, err)
|
||||
got, ok := models.LoadSessionRuntimeConfigOverrides(updated.Metadata)
|
||||
require.True(t, ok)
|
||||
require.Equal(t, overrides, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreflightWorkflowStepCredentials_SameProfileNew(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newProfileSwitchFixture(t, models.WorkflowProfileSessionStartPolicyNew, models.WorkflowProfileSessionEndPolicyPark)
|
||||
require.NoError(t, fixture.repo.CreateRepository(ctx, &models.Repository{
|
||||
ID: "repo-invalid", WorkspaceID: "ws1", Name: "widgets", SourceType: "local",
|
||||
Provider: "acme-forge", RemoteURL: "https://forge.example/acme/widgets.git",
|
||||
}))
|
||||
require.NoError(t, fixture.repo.CreateTaskRepository(ctx, &models.TaskRepository{
|
||||
ID: "taskrepo-invalid", TaskID: "t1", RepositoryID: "repo-invalid",
|
||||
}))
|
||||
fixture.svc.executor.SetGitHubCredentialBroker(
|
||||
fakeSwitchSessionCredentialIssuer{}, "https://kandev.example/api/v1/github/credentials/resolve",
|
||||
)
|
||||
|
||||
target := &wfmodels.WorkflowStep{
|
||||
ID: "step-b", WorkflowID: "wf1", Position: 1, AgentProfileID: "profile-a",
|
||||
ProfileSessionStartPolicy: models.WorkflowProfileSessionStartPolicyNew,
|
||||
}
|
||||
err := fixture.svc.preflightWorkflowStepCredentials(ctx, "t1", fixture.current, target)
|
||||
require.Error(t, err)
|
||||
|
||||
current, err := fixture.repo.GetTaskSession(ctx, fixture.current.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, models.TaskSessionStateRunning, current.State)
|
||||
require.True(t, current.IsPrimary)
|
||||
}
|
||||
|
||||
func TestProcessOnEnter_SameProfileNewPromptRecipient(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fixture := newProfileSwitchFixture(t, models.WorkflowProfileSessionStartPolicyNew, models.WorkflowProfileSessionEndPolicyPark)
|
||||
const marker = "same-profile destination marker"
|
||||
promptDelivered := make(chan struct{})
|
||||
processStarted := make(chan struct{})
|
||||
var promptOnce sync.Once
|
||||
var processOnce sync.Once
|
||||
var launchedSessionID, launchedPrompt string
|
||||
fixture.svc.executor.SetOnAgentProcessStarted(func(context.Context, string, string, string) {
|
||||
processOnce.Do(func() { close(processStarted) })
|
||||
})
|
||||
fixture.agentMgr.launchAgentFunc = func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
|
||||
if strings.Contains(req.TaskDescription, marker) {
|
||||
launchedSessionID = req.SessionID
|
||||
launchedPrompt = req.TaskDescription
|
||||
promptOnce.Do(func() { close(promptDelivered) })
|
||||
}
|
||||
return &executor.LaunchAgentResponse{AgentExecutionID: "execution-" + req.AgentProfileID}, nil
|
||||
}
|
||||
fixture.agentMgr.startAgentProcessFunc = func(_ context.Context, _ string) error {
|
||||
fixture.agentMgr.mu.Lock()
|
||||
for _, call := range fixture.agentMgr.setExecutionDescriptionCalls {
|
||||
if strings.Contains(call.Prompt, marker) {
|
||||
promptOnce.Do(func() { close(promptDelivered) })
|
||||
break
|
||||
}
|
||||
}
|
||||
fixture.agentMgr.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
target := &wfmodels.WorkflowStep{
|
||||
ID: "step-b", WorkflowID: "wf1", Position: 1, AgentProfileID: "profile-a",
|
||||
Prompt: marker,
|
||||
ProfileSessionStartPolicy: models.WorkflowProfileSessionStartPolicyNew,
|
||||
}
|
||||
source := &wfmodels.WorkflowStep{
|
||||
ID: "step-a", WorkflowID: "wf1", Position: 0, AgentProfileID: "profile-a",
|
||||
ProfileSessionEndPolicy: models.WorkflowProfileSessionEndPolicyPark,
|
||||
}
|
||||
|
||||
fixture.svc.processOnEnter(ctx, "t1", fixture.current, target, "task description", 0, source)
|
||||
select {
|
||||
case <-promptDelivered:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("same-profile new workflow entry did not deliver a prompt")
|
||||
}
|
||||
select {
|
||||
case <-processStarted:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("same-profile new workflow entry did not start the destination agent")
|
||||
}
|
||||
fixture.agentMgr.mu.Lock()
|
||||
descriptionCalls := append([]promptCall(nil), fixture.agentMgr.setExecutionDescriptionCalls...)
|
||||
fixture.agentMgr.mu.Unlock()
|
||||
if launchedSessionID != "" {
|
||||
require.NotEqual(t, fixture.current.ID, launchedSessionID)
|
||||
require.True(t, strings.Contains(launchedPrompt, marker))
|
||||
} else {
|
||||
markerIdx := -1
|
||||
for i, call := range descriptionCalls {
|
||||
if strings.Contains(call.Prompt, marker) {
|
||||
markerIdx = i
|
||||
break
|
||||
}
|
||||
}
|
||||
require.GreaterOrEqual(t, markerIdx, 0, "marker not found in setExecutionDescriptionCalls")
|
||||
require.NotEqual(t, "execution-a", descriptionCalls[markerIdx].ExecutionID)
|
||||
}
|
||||
|
||||
sessions, err := fixture.repo.ListTaskSessions(ctx, "t1")
|
||||
require.NoError(t, err)
|
||||
require.Len(t, sessions, 2)
|
||||
var destination *models.TaskSession
|
||||
for _, candidate := range sessions {
|
||||
if candidate.ID != fixture.current.ID && candidate.IsPrimary {
|
||||
destination = candidate
|
||||
break
|
||||
}
|
||||
}
|
||||
require.NotNil(t, destination)
|
||||
require.Equal(t, "profile-a", destination.AgentProfileID)
|
||||
require.NotEqual(t, fixture.current.ID, destination.ID)
|
||||
}
|
||||
@@ -258,6 +258,13 @@ type WorkflowStepGetter interface {
|
||||
GetNextStepByPosition(ctx context.Context, workflowID string, currentPosition int) (*wfmodels.WorkflowStep, error)
|
||||
}
|
||||
|
||||
// WorkflowMovePreflight validates the destination lifecycle before a task
|
||||
// move is committed. The orchestrator owns the credential and session-target
|
||||
// checks, while the task service owns the move transaction.
|
||||
type WorkflowMovePreflight interface {
|
||||
PreflightWorkflowStepMove(ctx context.Context, taskID string, currentSession *models.TaskSession, targetStep *wfmodels.WorkflowStep) error
|
||||
}
|
||||
|
||||
// workflowStepLister is an optional extension used to find WIP steps that
|
||||
// pull work from a feeder when new work arrives in that feeder.
|
||||
type workflowStepLister interface {
|
||||
@@ -424,6 +431,7 @@ type Service struct {
|
||||
workflowStepCreator WorkflowStepCreator
|
||||
workspaceBootstrapper WorkspaceBootstrapper
|
||||
workflowStepGetter WorkflowStepGetter
|
||||
workflowMovePreflight WorkflowMovePreflight
|
||||
startStepResolver StartStepResolver
|
||||
stepHistoryRecorder StepHistoryRecorder
|
||||
contributionDestinationPreparer ContributionDestinationPreparer
|
||||
@@ -715,6 +723,13 @@ func (s *Service) SetWorkflowStepGetter(getter WorkflowStepGetter) {
|
||||
s.workflowStepGetter = getter
|
||||
}
|
||||
|
||||
// SetWorkflowMovePreflight wires the orchestrator's synchronous destination
|
||||
// lifecycle validation for task moves. It is optional for standalone task
|
||||
// service users and tests.
|
||||
func (s *Service) SetWorkflowMovePreflight(preflight WorkflowMovePreflight) {
|
||||
s.workflowMovePreflight = preflight
|
||||
}
|
||||
|
||||
// SetStartStepResolver wires the start step resolver for CreateTask.
|
||||
func (s *Service) SetStartStepResolver(resolver StartStepResolver) {
|
||||
s.startStepResolver = resolver
|
||||
|
||||
@@ -611,6 +611,12 @@ func (s *Service) MoveTaskWithOptions(
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if stepChanged && targetStep != nil && s.workflowMovePreflight != nil {
|
||||
currentSession := s.resolvePrimaryOrActiveSession(ctx, id)
|
||||
if err := s.workflowMovePreflight.PreflightWorkflowStepMove(ctx, id, currentSession, targetStep); err != nil {
|
||||
return nil, fmt.Errorf("failed to preflight workflow move: %w", err)
|
||||
}
|
||||
}
|
||||
stateAfterAdmission := *task
|
||||
if stepChanged {
|
||||
if err := s.syncTaskStateForWorkflowMove(ctx, &stateAfterAdmission, oldStepID, workflowStepID, opts); err != nil {
|
||||
|
||||
@@ -58,6 +58,31 @@ func (testStepNotFound) Error() string { return "step not found" }
|
||||
|
||||
var errStepNotFoundForTest = testStepNotFound{}
|
||||
|
||||
type fakeWorkflowMovePreflight struct {
|
||||
err error
|
||||
calls int
|
||||
taskID string
|
||||
sessionID string
|
||||
targetStepID string
|
||||
}
|
||||
|
||||
func (f *fakeWorkflowMovePreflight) PreflightWorkflowStepMove(
|
||||
_ context.Context,
|
||||
taskID string,
|
||||
currentSession *models.TaskSession,
|
||||
targetStep *wfmodels.WorkflowStep,
|
||||
) error {
|
||||
f.calls++
|
||||
f.taskID = taskID
|
||||
if currentSession != nil {
|
||||
f.sessionID = currentSession.ID
|
||||
}
|
||||
if targetStep != nil {
|
||||
f.targetStepID = targetStep.ID
|
||||
}
|
||||
return f.err
|
||||
}
|
||||
|
||||
// TestService_SetWorkflowHidden_HealsStaleRecord verifies the helper used by
|
||||
// the improve-kandev bootstrap to flip Hidden=true on workflows created
|
||||
// before the flag was honored on insert.
|
||||
@@ -237,6 +262,46 @@ func TestService_MoveTaskAllowsPendingReviewWhenSessionIdle(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestService_MoveTaskPreflightsWorkflowLifecycleBeforeCommit(t *testing.T) {
|
||||
svc, eventBus, repo := createTestService(t)
|
||||
ctx := context.Background()
|
||||
seedMoveWorkflows(t, ctx, repo)
|
||||
seedMoveSteps(svc)
|
||||
createMoveTask(t, ctx, repo, "task-preflight", "wf-source", "step-source", nil)
|
||||
createMoveSession(t, ctx, repo, "session-preflight", "task-preflight", models.TaskSessionStateRunning, models.ReviewStatusNone)
|
||||
preflightErr := errors.New("managed credentials are invalid")
|
||||
preflight := &fakeWorkflowMovePreflight{err: preflightErr}
|
||||
svc.SetWorkflowMovePreflight(preflight)
|
||||
|
||||
_, err := svc.MoveTaskWithOptions(ctx, "task-preflight", "wf-source", "step-review-target", 0, MoveTaskOptions{
|
||||
AllowActivePrimarySession: true,
|
||||
})
|
||||
if !errors.Is(err, preflightErr) {
|
||||
t.Fatalf("MoveTaskWithOptions error = %v, want %v", err, preflightErr)
|
||||
}
|
||||
if preflight.calls != 1 {
|
||||
t.Fatalf("workflow move preflight calls = %d, want 1", preflight.calls)
|
||||
}
|
||||
if preflight.taskID != "task-preflight" || preflight.sessionID != "session-preflight" || preflight.targetStepID != "step-review-target" {
|
||||
t.Fatalf("preflight inputs = (%q, %q, %q), want (%q, %q, %q)",
|
||||
preflight.taskID, preflight.sessionID, preflight.targetStepID,
|
||||
"task-preflight", "session-preflight", "step-review-target")
|
||||
}
|
||||
|
||||
task, err := repo.GetTask(ctx, "task-preflight")
|
||||
if err != nil {
|
||||
t.Fatalf("GetTask: %v", err)
|
||||
}
|
||||
if task.WorkflowStepID != "step-source" {
|
||||
t.Fatalf("task step = %q, want source step after preflight failure", task.WorkflowStepID)
|
||||
}
|
||||
for _, event := range eventBus.GetPublishedEvents() {
|
||||
if event.Type == events.TaskMoved {
|
||||
t.Fatal("task.moved published after preflight failure")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestService_MoveTaskToTerminalStepCompletesTask(t *testing.T) {
|
||||
svc, eventBus, repo := createTestService(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { test, expect } from "../../fixtures/test-base";
|
||||
import { SessionPage } from "../../pages/session-page";
|
||||
import { WorkflowSettingsPage } from "../../pages/workflow-settings-page";
|
||||
import {
|
||||
createWorkflowAgentProfiles,
|
||||
@@ -171,6 +172,93 @@ test.describe("Workflow session targeting", () => {
|
||||
);
|
||||
});
|
||||
|
||||
test("honors same-profile new sessions from the task topbar", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
test.setTimeout(120_000);
|
||||
const { profileA } = await createWorkflowAgentProfiles(apiClient);
|
||||
const workflow = await apiClient.createWorkflow(
|
||||
seedData.workspaceId,
|
||||
"Same Profile Topbar Session",
|
||||
);
|
||||
const source = await apiClient.createWorkflowStep(workflow.id, "Source", 0, {
|
||||
agent_profile_id: profileA.id,
|
||||
profile_session_end_policy: "park",
|
||||
});
|
||||
const destination = await apiClient.createWorkflowStep(workflow.id, "Destination", 1, {
|
||||
agent_profile_id: profileA.id,
|
||||
profile_session_start_policy: "new",
|
||||
profile_session_end_policy: "park",
|
||||
});
|
||||
const marker = "same-profile-new-topbar";
|
||||
await apiClient.updateWorkflowStep(destination.id, {
|
||||
prompt: `e2e:message("${marker}")`,
|
||||
});
|
||||
|
||||
const task = await apiClient.createTaskWithAgent(
|
||||
seedData.workspaceId,
|
||||
"Same profile topbar session",
|
||||
profileA.id,
|
||||
{
|
||||
workflow_id: workflow.id,
|
||||
workflow_step_id: source.id,
|
||||
repository_ids: [seedData.repositoryId],
|
||||
},
|
||||
);
|
||||
const sourceSessionId = await waitForWorkflowProfileSession(apiClient, task.id, profileA.id);
|
||||
|
||||
await testPage.goto(`/t/${task.id}`);
|
||||
const session = new SessionPage(testPage);
|
||||
await session.waitForLoad();
|
||||
await session.moveToWorkflowStep({ id: destination.id, name: "Destination" });
|
||||
|
||||
let destinationSessionId = "";
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
const { sessions } = await apiClient.listTaskSessions(task.id);
|
||||
const destinationSession = sessions.find(
|
||||
(candidate) => candidate.id !== sourceSessionId && candidate.is_primary,
|
||||
);
|
||||
destinationSessionId = destinationSession?.id ?? "";
|
||||
return destinationSessionId;
|
||||
},
|
||||
{ timeout: 30_000, message: "same-profile topbar move did not create a primary session" },
|
||||
)
|
||||
.not.toBe("");
|
||||
|
||||
await expect
|
||||
.poll(async () => (await apiClient.getTask(task.id)).workflow_step_id, {
|
||||
timeout: 15_000,
|
||||
})
|
||||
.toBe(destination.id);
|
||||
await waitForAgentMarker(apiClient, destinationSessionId, marker);
|
||||
|
||||
const movedSessions = (await apiClient.listTaskSessions(task.id)).sessions;
|
||||
expect(movedSessions).toHaveLength(2);
|
||||
expect(movedSessions.find((candidate) => candidate.id === destinationSessionId)).toMatchObject({
|
||||
agent_profile_id: profileA.id,
|
||||
is_primary: true,
|
||||
});
|
||||
expect(movedSessions.find((candidate) => candidate.id === sourceSessionId)?.state).toBe(
|
||||
"WAITING_FOR_INPUT",
|
||||
);
|
||||
await expect(session.activeChat()).toContainText(marker, { timeout: 15_000 });
|
||||
|
||||
await testPage.reload();
|
||||
await session.waitForLoad();
|
||||
const reloadedSessions = (await apiClient.listTaskSessions(task.id)).sessions;
|
||||
expect(
|
||||
reloadedSessions.find((candidate) => candidate.id === destinationSessionId),
|
||||
).toMatchObject({
|
||||
agent_profile_id: profileA.id,
|
||||
is_primary: true,
|
||||
});
|
||||
await expect(session.activeChat()).toContainText(marker, { timeout: 15_000 });
|
||||
});
|
||||
|
||||
test("delivers a source-step target to its recorded conversation", async ({
|
||||
apiClient,
|
||||
seedData,
|
||||
|
||||
@@ -32,10 +32,17 @@ The defaults are `reuse` and `complete`. These defaults preserve existing
|
||||
behavior. The router reuses an eligible nonterminal destination session when one
|
||||
exists. It completes the source session.
|
||||
|
||||
For profile-only routing, the settings apply only when the effective agent
|
||||
profile changes. Consecutive steps with the same effective profile keep the
|
||||
active session. The proposed explicit-target extension below distinguishes
|
||||
session identity from profile identity.
|
||||
For profile-only routing, consecutive steps with the same effective profile
|
||||
keep the active session when the destination uses `reuse`. A destination with
|
||||
`new` creates a fresh conversation even when profile IDs match. The source
|
||||
step's end policy applies when that conversation is replaced.
|
||||
|
||||
This paragraph was reconciled on 2026-09-11 with existing acceptance criterion
|
||||
AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.5. Profile comparison uses logical profile
|
||||
IDs. Reuse preserves user-selected runtime model and reasoning overrides; it
|
||||
does not compare or restore model settings. The
|
||||
[repair package](../plans/workflow-same-profile-new-session/plan.md) tracks the
|
||||
remaining implementation gap.
|
||||
|
||||
The workflow engine keeps its transition graph, actions, events, and states. The
|
||||
transition integration must provide both source and destination settings to the
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
---
|
||||
created: 2026-09-11
|
||||
status: complete
|
||||
requirements:
|
||||
- REQ-TASKS-WORKFLOW-PROFILE-SESSIONS-001
|
||||
system_design:
|
||||
- ../../specs/tasks/system-design/workflow-profile-session-lifecycle.md
|
||||
legacy_specs: []
|
||||
---
|
||||
|
||||
# Implementation Plan: Same-profile fresh workflow sessions
|
||||
|
||||
## Overview
|
||||
|
||||
Honor the destination's `new` policy when its resolved profile matches the
|
||||
active session. One sequential work order corrects routing and verifies the
|
||||
manual-move outcome. The task system owns this change because it selects the
|
||||
conversation that receives workflow entry actions.
|
||||
|
||||
## Evidence and requirement conformance
|
||||
|
||||
At checkout `407ed4f1a5`, `prepareWorkflowStepSession` returns the active session
|
||||
when profile IDs match, before it reads `ProfileSessionStartPolicy`.
|
||||
`preflightWorkflowStepCredentials` has the same shortcut. This violates
|
||||
[AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.2 and 001.5](../../specs/tasks/requirements/workflow-profile-session-lifecycle.md).
|
||||
The existing requirement and design already specify the correction; their draft
|
||||
status also covers a broader recipient feature and is not changed by this repair.
|
||||
|
||||
The reported live task is compatibility evidence, not a reproduction of this
|
||||
defect. Its saved destination policy was `reuse`. Its session retained the Luna
|
||||
profile ID after a user selected Astra as a runtime override. Reusing that
|
||||
session was correct. Profile identity remains the routing comparison; runtime
|
||||
model and reasoning overrides do not trigger a switch or a reset.
|
||||
|
||||
Minimal reproduction: create a task session with profile A, then enter a second
|
||||
step with profile A and start policy `new`. The current shortcut returns the
|
||||
original session instead of a fresh conversation. Use a focused service test
|
||||
for RED; no failing test has been executed during package creation.
|
||||
|
||||
## Scope
|
||||
|
||||
### In scope
|
||||
|
||||
- Profile-only entry with the same nonempty resolved profile and `new`.
|
||||
- Consistent credential preflight and source `park`/`complete` handling.
|
||||
- Same-profile `reuse` and default reuse preserving runtime overrides.
|
||||
- Entry prompt delivery to the replacement and its promotion to primary.
|
||||
|
||||
### Out of scope
|
||||
|
||||
- Comparing effective models, resetting overrides, or changing agent profiles.
|
||||
- Changing empty-profile fallback, initial creation, or explicit recipient rules.
|
||||
- New settings, UI markup, translations, schema, APIs, or Office behavior.
|
||||
- Reworking routing persistence or unrelated retry behavior.
|
||||
|
||||
## Technical approach
|
||||
|
||||
In `apps/backend/internal/orchestrator/event_handlers_workflow.go`, resolve the
|
||||
destination start policy before the same-profile return in
|
||||
`prepareWorkflowStepSession`. Keep the empty-profile fallback. Keep a matching
|
||||
profile only for normalized `reuse`; send matching-profile `new` through
|
||||
`switchSessionForStepWithPolicies` with the existing source end policy.
|
||||
Apply the same decision in `preflightWorkflowStepCredentials` so replacement
|
||||
validation cannot be skipped. Do not add model comparisons.
|
||||
|
||||
Use the existing replacement, promotion, queue transfer, source-binding, and
|
||||
stamped runtime-stop paths. Preserve the explicit-target branch in
|
||||
`workflow_session_target.go`. No source session may be retired before a valid
|
||||
destination is prepared. A replacement keeps the task environment and executor
|
||||
inheritance, and starts its own provider conversation.
|
||||
|
||||
The task service invokes the same destination credential preflight before
|
||||
committing a service-level workflow move. This keeps an asynchronous
|
||||
`task.moved` lifecycle failure from leaving the task persisted on a step with
|
||||
no prepared destination session.
|
||||
|
||||
## Tests
|
||||
|
||||
Add `event_handlers_workflow_same_profile_policy_test.go` beside the existing
|
||||
profile-policy fixtures. Planned test names and evidence:
|
||||
|
||||
| Test | Evidence | Criteria |
|
||||
| --- | --- | --- |
|
||||
| `TestPrepareWorkflowStepSession_SameProfileNew` | Distinct session with profile A, no reuse of another matching session; park/complete table cases | 001.2, 001.4, 001.5 |
|
||||
| `TestPrepareWorkflowStepSession_SameProfileReusePreservesOverrides` | Same ID and metadata for explicit/default reuse, even with another model override | 001.5 |
|
||||
| `TestPreflightWorkflowStepCredentials_SameProfileNew` | Credential rejection preserves source and prevents destination prompt | 001.11 |
|
||||
| `TestProcessOnEnter_SameProfileNewPromptRecipient` | Entry actions reach the fresh primary; source has no destination prompt | 001.2, 001.5 |
|
||||
|
||||
Use `newProfileSwitchFixture` and real repository session rows. Capture launch
|
||||
and prompt recipients rather than asserting only that a helper was called.
|
||||
Retain existing different-profile and explicit-target regressions.
|
||||
|
||||
## E2E tests
|
||||
|
||||
Extend `apps/web/e2e/tests/workflow/workflow-session-targeting.spec.ts` in project
|
||||
`chromium` with `same-profile new session from topbar`. Seed a disposable
|
||||
workflow using the same profile for source and destination, destination `new`,
|
||||
source `park`, and a unique destination prompt marker. Move through the task
|
||||
topbar. Assert a distinct primary session, visible replacement conversation,
|
||||
and marker delivery to that conversation. Reload and confirm the primary
|
||||
selection persists. Use API state to corroborate identities and source parking.
|
||||
This covers 001.2, 001.4, and 001.5 through the reported entry point.
|
||||
|
||||
No UI structure changes are planned, so no UI preview is required. Existing
|
||||
mobile recipient tests remain compatibility evidence; this package changes
|
||||
the shared backend policy rather than phone interaction.
|
||||
|
||||
## Work orders
|
||||
|
||||
- [x] [Task 01: Honor same-profile new-session policy](task-01-honor-new-session-policy.md)
|
||||
|
||||
## Related delivery records
|
||||
|
||||
[Workflow session targeting](../workflow-session-targeting/plan.md) delivered
|
||||
explicit initial and source-step recipients. Its completed results remain
|
||||
historical evidence, not proof of this profile-only case. This package owns
|
||||
the missing case and does not reopen its eight work orders.
|
||||
|
||||
## Verification results
|
||||
|
||||
Implementation and package validation on 2026-09-11:
|
||||
|
||||
- RED: `go test ./internal/orchestrator -run '^TestPrepareWorkflowStepSession_SameProfileNew$' -count=1` failed all three table cases before the production change because the matching-profile shortcut returned `switched=false`.
|
||||
- GREEN: the focused same-profile preparation, reuse, credential preflight, and entry-prompt tests passed 8 cases.
|
||||
- `go test -race ./internal/orchestrator -run 'TestPrepareWorkflowStepSession|TestPreflightWorkflowStepCredentials|TestProcessOnEnter|TestSwitchSessionForStep|TestWorkflowSessionTarget' -count=1`: 99 passed with no race reports.
|
||||
- `make -C apps/backend build`: passed.
|
||||
- `make -C apps/backend lint`: passed with 0 issues.
|
||||
- `pnpm e2e:run --project chromium tests/workflow/workflow-session-targeting.spec.ts`: 5 passed with a fresh managed build.
|
||||
- `pnpm e2e:run --no-build --project chromium tests/workflow/workflow-session-targeting.spec.ts`: 5 passed after refreshing the E2E plugin artifact.
|
||||
- `pnpm run typecheck`: passed.
|
||||
- `python3 scripts/lint-spec-files.py --all`: passed.
|
||||
- `git diff --check`: passed.
|
||||
- `gofmt -l` and the targeted Prettier check: passed.
|
||||
|
||||
## Risks
|
||||
|
||||
- A policy-blind preflight could skip credential validation for new sessions.
|
||||
- Broader equality changes could reset user overrides or alter explicit targets.
|
||||
- Wrong source-step propagation could complete a session intended to be parked.
|
||||
- A topbar assertion based only on labels could miss wrong-session delivery.
|
||||
|
||||
## Documentation impact
|
||||
|
||||
Public workflow documentation already says `new` always starts a fresh
|
||||
conversation. No public copy change is needed. Reconcile the older ADR's
|
||||
same-profile exception with the existing requirement; do not change defaults
|
||||
or unrelated historical decisions in this package.
|
||||
@@ -0,0 +1,129 @@
|
||||
---
|
||||
id: "01-honor-new-session-policy"
|
||||
title: "Honor same-profile new-session policy"
|
||||
status: complete
|
||||
wave: 1
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
requirements:
|
||||
- REQ-TASKS-WORKFLOW-PROFILE-SESSIONS-001
|
||||
acceptance_criteria:
|
||||
- AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.2
|
||||
- AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.4
|
||||
- AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.5
|
||||
- AC-TASKS-WORKFLOW-PROFILE-SESSIONS-001.11
|
||||
system_design:
|
||||
- ../../specs/tasks/system-design/workflow-profile-session-lifecycle.md
|
||||
---
|
||||
|
||||
# Task 01: Honor same-profile new-session policy
|
||||
|
||||
## Summary
|
||||
|
||||
Apply the destination start policy before returning a matching-profile session.
|
||||
Keep profile-ID comparison and preserve overrides whenever reuse is selected.
|
||||
|
||||
## In scope
|
||||
|
||||
- Add the four backend regressions named in the plan; run the `new` case RED
|
||||
before changing production code. Test both source end policies and seed an
|
||||
extra matching session to prove `new` does not choose it.
|
||||
- Correct preparation and credential preflight with the same policy decision.
|
||||
- Exercise `processOnEnter` prompt recipient and primary-session promotion.
|
||||
- Add the topbar E2E case from the plan using disposable mock-agent fixtures.
|
||||
- Record actual targeted results and update this work order and plan status.
|
||||
|
||||
## Out of scope
|
||||
|
||||
Model-based routing, runtime override resets, no-profile fallback changes,
|
||||
explicit-target redesign, UI changes, database migrations, and live-task edits.
|
||||
|
||||
## Acceptance
|
||||
|
||||
1. Matching nonempty profiles plus `new` create a distinct conversation using
|
||||
the selected profile and existing environment. Source `park` and `complete`
|
||||
outcomes remain correct. The replacement receives the step prompt.
|
||||
2. Matching profiles plus explicit/default `reuse` retain the same session and
|
||||
runtime overrides. Different-profile and explicit-target tests still pass.
|
||||
3. Credential/preparation failure leaves the source recoverable without a
|
||||
destination prompt. The topbar scenario persists the new primary on reload.
|
||||
|
||||
## Verification
|
||||
|
||||
Run from the repository root. Install workspace dependencies once if absent.
|
||||
The focused Go commands are used because the Make test target runs all packages.
|
||||
|
||||
```bash
|
||||
(cd apps && rtk pnpm install --frozen-lockfile)
|
||||
(cd apps/backend && rtk go test ./internal/orchestrator -run '^TestPrepareWorkflowStepSession_SameProfileNew$' -count=1)
|
||||
(cd apps/backend && rtk go test -race ./internal/orchestrator -run 'TestPrepareWorkflowStepSession|TestPreflightWorkflowStepCredentials|TestProcessOnEnter|TestSwitchSessionForStep|TestWorkflowSessionTarget' -count=1)
|
||||
(cd apps/web && rtk pnpm e2e:run --project chromium tests/workflow/workflow-session-targeting.spec.ts)
|
||||
rtk proxy python3 scripts/lint-spec-files.py --all
|
||||
rtk git diff --check
|
||||
```
|
||||
|
||||
First run the new focused test before the fix and record the expected assertion
|
||||
failure. Run it again after the fix, then the remaining checks. Use a fresh
|
||||
managed E2E build. Do not treat missing tests or fixture failure as RED evidence.
|
||||
Use causal waits and existing session/page helpers; do not add timed sleeps.
|
||||
|
||||
## Files likely touched
|
||||
|
||||
- `apps/backend/internal/orchestrator/event_handlers_workflow.go`
|
||||
- `apps/backend/internal/orchestrator/event_handlers_workflow_same_profile_policy_test.go` (new)
|
||||
- `apps/web/e2e/tests/workflow/workflow-session-targeting.spec.ts`
|
||||
- This package's plan and work order for status and results.
|
||||
|
||||
## Dependencies
|
||||
|
||||
None. Work is sequential in the primary session after an explicit implementation request.
|
||||
|
||||
## Risks
|
||||
|
||||
Use the existing profile-switch fixtures and failure injection. Avoid treating
|
||||
a source model override as another profile. Keep source-step validation and
|
||||
the replacement-before-retirement order intact.
|
||||
|
||||
## Parallelism
|
||||
|
||||
`sequential`
|
||||
|
||||
## Inputs
|
||||
|
||||
- [Requirements](../../specs/tasks/requirements/workflow-profile-session-lifecycle.md), criteria listed above.
|
||||
- [Design](../../specs/tasks/system-design/workflow-profile-session-lifecycle.md), Control flow and Failure and recovery.
|
||||
- `event_handlers_workflow_profile_session_policy_test.go`: `newProfileSwitchFixture` and independent start/end policy tests.
|
||||
- `event_handlers_workflow_profile_test.go`: existing same-profile reuse test.
|
||||
- `workflow_session_target.go`: explicit-target compatibility path.
|
||||
- [Plan](plan.md): root cause, test names, topbar scenario, and exclusions.
|
||||
|
||||
## Results
|
||||
|
||||
Implemented the policy-aware profile-only routing path. The preparation and
|
||||
credential preflight paths now resolve the destination start policy before
|
||||
their same-profile shortcut. Matching profiles preserve the current session
|
||||
only for normalized `reuse` or an empty profile; matching `new` uses the
|
||||
existing replacement lifecycle and the source end policy. Runtime model and
|
||||
reasoning overrides remain untouched.
|
||||
|
||||
The task service now runs the orchestrator's destination credential preflight
|
||||
before committing a service-level workflow move. A failed preflight therefore
|
||||
leaves the source step unchanged instead of relying on the asynchronous
|
||||
`task.moved` lifecycle handler to reject the destination.
|
||||
|
||||
Added backend regressions for same-profile `new` with both source end policies,
|
||||
explicit/default `reuse` override preservation, credential preflight failure,
|
||||
and `processOnEnter` prompt delivery to the replacement. Added the Chromium
|
||||
topbar E2E regression with API identity checks and reload persistence.
|
||||
|
||||
TDD and verification results:
|
||||
|
||||
- RED focused test: 3 table cases failed before the production change because
|
||||
the same-profile shortcut returned `switched=false`.
|
||||
- GREEN focused policy tests: 8 passed.
|
||||
- Race-filtered orchestrator tests: 99 passed with no race reports.
|
||||
- Backend build and lint: passed; lint reported 0 issues.
|
||||
- Workflow session-targeting Chromium E2E: 5 passed after the final backend
|
||||
build and E2E plugin packaging.
|
||||
- Web typecheck, specification lint, Go formatting, targeted Prettier check,
|
||||
and `git diff --check`: passed.
|
||||
@@ -352,6 +352,10 @@ run but reported no matching tests in the current checkout.
|
||||
|
||||
## Risks
|
||||
|
||||
Follow-up: [same-profile fresh-session repair](../workflow-same-profile-new-session/plan.md)
|
||||
covers the profile-only `new` shortcut discovered on 2026-09-11. The completed
|
||||
explicit-target results above do not establish coverage for that case.
|
||||
|
||||
- Existing same-profile early returns can bypass explicit recipient intent.
|
||||
- Initial provenance must survive primary changes without acquiring new markers.
|
||||
- A step binding written before a failed promotion could misroute later work.
|
||||
|
||||
@@ -184,7 +184,8 @@ The desktop popover has this hierarchy:
|
||||
generic icon and explains that task creation determines the agent.
|
||||
3. The lifecycle view explains that the start setting selects the target's
|
||||
conversation and the end setting applies when leaving that conversation.
|
||||
Profile-only choices retain their existing same-profile behavior.
|
||||
Profile-only choices preserve the current session for same-profile `reuse`;
|
||||
`new` replaces it even when profile IDs match.
|
||||
4. **When this step starts** offers:
|
||||
- **Reuse an available session.** Continue the most recent available session
|
||||
for this target. If none is available, start a new session. For an explicit
|
||||
@@ -484,6 +485,11 @@ Profile-switch logs include source step ID, destination step ID, start setting,
|
||||
end setting, source outcome, and destination outcome. Stop-event suppression
|
||||
logs include session ID, execution ID, and intent stamp.
|
||||
|
||||
## Implementation plans
|
||||
|
||||
- [Explicit session targeting](../../../plans/workflow-session-targeting/plan.md)
|
||||
- [Same-profile fresh-session repair](../../../plans/workflow-same-profile-new-session/plan.md)
|
||||
|
||||
## Related decisions
|
||||
|
||||
- [Make workflow step profile-session switching explicit](../../../decisions/2026-08-31-workflow-profile-session-switch-policy.md)
|
||||
|
||||
Reference in New Issue
Block a user