fix: bind office run identity to its own agent's session, not the assignee's (#3211)
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:
@@ -1276,6 +1276,7 @@ func initOfficeServices(
|
||||
agentRegistry, log, services, cfg.Office.JWTSigningKey,
|
||||
)
|
||||
wireOfficeSvcsDependencies(services, repos, eventBus, orchestratorSvc, agentRegistry)
|
||||
services.OfficeSvcs.Dashboard.SetOfficeSessionIdentity(cfg.Features.OfficeSessionIdentity)
|
||||
|
||||
// Reconcile using the new infra package.
|
||||
reconciler := officeinfra.NewReconciler(repos.Office, log)
|
||||
|
||||
@@ -82,6 +82,8 @@ func provideOrchestrator(
|
||||
cfg != nil && cfg.Features.ClaudeBackgroundPromptHandoff
|
||||
serviceCfg.ClaudeMidTurnSteering =
|
||||
cfg != nil && cfg.Features.ClaudeMidTurnSteering
|
||||
serviceCfg.OfficeSessionIdentity =
|
||||
cfg != nil && cfg.Features.OfficeSessionIdentity
|
||||
namespace := resolveEventNamespace(cfg)
|
||||
serviceCfg.QueueGroup = "orchestrator." + namespace
|
||||
busMode := "memory"
|
||||
|
||||
@@ -130,6 +130,7 @@ var startupExclusions = []CatalogExclusion{
|
||||
{EnvVar: "KANDEV_FEATURES_AUTH", Class: "profile", Reason: "runtime feature flag registry"},
|
||||
{EnvVar: "KANDEV_FEATURES_CLAUDE_BACKGROUND_PROMPT_HANDOFF", Class: "debug", Reason: "runtime feature flag registry"},
|
||||
{EnvVar: "KANDEV_FEATURES_CLAUDE_MID_TURN_STEERING", Class: "debug", Reason: "runtime feature flag registry"},
|
||||
{EnvVar: "KANDEV_FEATURES_OFFICE_SESSION_IDENTITY", Class: "debug", Reason: "runtime feature flag registry"},
|
||||
{EnvVar: "KANDEV_DEBUG_AGENT_MESSAGES", Class: "debug", Reason: "ACP frame diagnostics"},
|
||||
{EnvVar: "KANDEV_DEBUG_ACP_MAX_FILES", Class: "debug", Reason: "ACP debug retention"},
|
||||
{EnvVar: "KANDEV_DEBUG_ACP_RETENTION_HOURS", Class: "debug", Reason: "ACP debug retention"},
|
||||
|
||||
@@ -150,6 +150,7 @@ func auditedStartupEnvironmentInventory() []auditedStartupEnvironment {
|
||||
{envVar: "KANDEV_FEATURES_AUTH", class: "exclusion"},
|
||||
{envVar: "KANDEV_FEATURES_CLAUDE_BACKGROUND_PROMPT_HANDOFF", class: "exclusion"},
|
||||
{envVar: "KANDEV_FEATURES_CLAUDE_MID_TURN_STEERING", class: "exclusion"},
|
||||
{envVar: "KANDEV_FEATURES_OFFICE_SESSION_IDENTITY", class: "exclusion"},
|
||||
{envVar: "KANDEV_DEBUG_AGENT_MESSAGES", class: "exclusion"},
|
||||
{envVar: "KANDEV_DEBUG_ACP_MAX_FILES", class: "exclusion"},
|
||||
{envVar: "KANDEV_DEBUG_ACP_RETENTION_HOURS", class: "exclusion"},
|
||||
|
||||
@@ -461,6 +461,14 @@ type FeaturesConfig struct {
|
||||
// kill-switch after rollout because the agent-side fold is undocumented and
|
||||
// can regress without notice.
|
||||
ClaudeMidTurnSteering bool `mapstructure:"claude_mid_turn_steering" json:"claudeMidTurnSteering"`
|
||||
|
||||
// OfficeSessionIdentity keys an Office task's session identity on the run's
|
||||
// own agent instead of the task's runner seat, and binds an agent's
|
||||
// decision re-evaluation to its own calling session instead of the task's
|
||||
// most-recently-started session. Off in every embedded profile: enabling it
|
||||
// exposes pre-existing duplicate (task_id, agent_profile_id) rows until the
|
||||
// companion unique-index fix has shipped.
|
||||
OfficeSessionIdentity bool `mapstructure:"office_session_identity" json:"officeSessionIdentity"`
|
||||
}
|
||||
|
||||
// LoggingConfig holds logging configuration.
|
||||
|
||||
@@ -63,6 +63,7 @@ func (h *Handlers) handleRecordStepDecision(ctx context.Context, msg *ws.Message
|
||||
AgentProfileID: session.AgentProfileID,
|
||||
Decision: req.Decision,
|
||||
Reason: req.Reason,
|
||||
SessionID: req.SessionID,
|
||||
})
|
||||
if err != nil {
|
||||
if !errors.Is(err, shared.ErrForbidden) && !dashboard.IsAgentDecisionValidationError(err) {
|
||||
|
||||
@@ -25,6 +25,13 @@ type RecordAgentDecisionInput struct {
|
||||
AgentProfileID string
|
||||
Decision string
|
||||
Reason string
|
||||
// SessionID, when features.officeSessionIdentity is on, names the
|
||||
// decider's own calling session so RecordDecision re-evaluates against
|
||||
// it instead of the task's most-recently-started ("active") session.
|
||||
// Populated unconditionally by the MCP handler; gated here because the
|
||||
// flag decision belongs with the rest of this service's behavior, not
|
||||
// the transport layer.
|
||||
SessionID string
|
||||
}
|
||||
|
||||
// AgentDecisionValidationError marks an error caused by a caller-provided
|
||||
@@ -137,7 +144,7 @@ func (s *DashboardService) RecordAgentDecision(
|
||||
return nil, agentDecisionValidation(fmt.Errorf("%s", agentDecisionReasonRequiredErr))
|
||||
}
|
||||
|
||||
result, err := dispatcher.RecordDecision(ctx, officeenginedispatcher.RecordDecisionInput{
|
||||
decisionInput := officeenginedispatcher.RecordDecisionInput{
|
||||
TaskID: in.TaskID,
|
||||
StepID: stepID,
|
||||
ParticipantID: participantID,
|
||||
@@ -146,7 +153,11 @@ func (s *DashboardService) RecordAgentDecision(
|
||||
DeciderID: in.AgentProfileID,
|
||||
Role: role,
|
||||
Comment: in.Reason,
|
||||
})
|
||||
}
|
||||
if s.officeSessionIdentity {
|
||||
decisionInput.SessionID = in.SessionID
|
||||
}
|
||||
result, err := dispatcher.RecordDecision(ctx, decisionInput)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("record decision: %w", err)
|
||||
}
|
||||
|
||||
@@ -6,11 +6,106 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/office/dashboard"
|
||||
officeenginedispatcher "github.com/kandev/kandev/internal/office/engine_dispatcher"
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
"github.com/kandev/kandev/internal/workflow/engine"
|
||||
)
|
||||
|
||||
// spyDecisionDispatcher satisfies shared.WorkflowEngineDispatcher plus the
|
||||
// package-local decisionRecordingDispatcher/roleResolvingDispatcher/
|
||||
// quorumEvaluatingDispatcher capabilities RecordAgentDecision reaches via
|
||||
// type assertion (all exported methods, so a type from this external test
|
||||
// package can still satisfy those unexported interfaces structurally).
|
||||
// Unlike newTestEngineDispatcher's real engine, this spy captures the exact
|
||||
// RecordDecisionInput it was called with, so a test can assert whether
|
||||
// SessionID was forwarded without needing a live session row to observe an
|
||||
// engine-side behavior difference.
|
||||
type spyDecisionDispatcher struct {
|
||||
role string
|
||||
participantID string
|
||||
lastInput officeenginedispatcher.RecordDecisionInput
|
||||
}
|
||||
|
||||
func (s *spyDecisionDispatcher) HandleTrigger(
|
||||
_ context.Context, _ string, _ engine.Trigger, _ any, _ string,
|
||||
) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *spyDecisionDispatcher) ResolveParticipantRole(
|
||||
_ context.Context, _, _, _ string,
|
||||
) (string, string, error) {
|
||||
return s.role, s.participantID, nil
|
||||
}
|
||||
|
||||
func (s *spyDecisionDispatcher) RecordDecision(
|
||||
_ context.Context, in officeenginedispatcher.RecordDecisionInput,
|
||||
) (officeenginedispatcher.RecordDecisionResult, error) {
|
||||
s.lastInput = in
|
||||
return officeenginedispatcher.RecordDecisionResult{DecisionID: "decision-spy"}, nil
|
||||
}
|
||||
|
||||
func (s *spyDecisionDispatcher) EvaluateStepQuorum(
|
||||
_ context.Context, _ string,
|
||||
) (engine.QuorumSnapshot, error) {
|
||||
return engine.QuorumSnapshot{}, nil
|
||||
}
|
||||
|
||||
// TestRecordAgentDecision_SessionIDNotForwardedWhenFlagOff is the default
|
||||
// (flag-off) case: even though the caller supplies a SessionID, the
|
||||
// dispatcher never sees it, so RecordDecision falls back to its existing
|
||||
// resolveActiveSessionID path — the pre-office-session-identity behavior.
|
||||
func TestRecordAgentDecision_SessionIDNotForwardedWhenFlagOff(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "sid1", "ws-d", "SID1", "in_review", 2)
|
||||
mustAddParticipant(t, deps, "sid1", "agent-1", models.ParticipantRoleApprover)
|
||||
spy := &spyDecisionDispatcher{role: models.ParticipantRoleApprover, participantID: "participant-1"}
|
||||
deps.svc.SetWorkflowEngineDispatcher(spy)
|
||||
|
||||
_, err := deps.svc.RecordAgentDecision(context.Background(), dashboard.RecordAgentDecisionInput{
|
||||
TaskID: "sid1",
|
||||
AgentProfileID: "agent-1",
|
||||
Decision: engine.DecisionApproved,
|
||||
Reason: "lgtm",
|
||||
SessionID: "sess-caller",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordAgentDecision: %v", err)
|
||||
}
|
||||
if spy.lastInput.SessionID != "" {
|
||||
t.Errorf("SessionID = %q, want blank (flag is off)", spy.lastInput.SessionID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordAgentDecision_SessionIDForwardedWhenFlagOn proves that once
|
||||
// features.officeSessionIdentity is enabled, RecordAgentDecision forwards
|
||||
// the caller's own SessionID into RecordDecisionInput, so RecordDecision
|
||||
// re-evaluates against the decider's own session instead of the task's
|
||||
// most-recently-started ("active") one.
|
||||
func TestRecordAgentDecision_SessionIDForwardedWhenFlagOn(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "sid2", "ws-d", "SID2", "in_review", 2)
|
||||
mustAddParticipant(t, deps, "sid2", "agent-1", models.ParticipantRoleApprover)
|
||||
spy := &spyDecisionDispatcher{role: models.ParticipantRoleApprover, participantID: "participant-1"}
|
||||
deps.svc.SetWorkflowEngineDispatcher(spy)
|
||||
deps.svc.SetOfficeSessionIdentity(true)
|
||||
|
||||
_, err := deps.svc.RecordAgentDecision(context.Background(), dashboard.RecordAgentDecisionInput{
|
||||
TaskID: "sid2",
|
||||
AgentProfileID: "agent-1",
|
||||
Decision: engine.DecisionApproved,
|
||||
Reason: "lgtm",
|
||||
SessionID: "sess-caller",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordAgentDecision: %v", err)
|
||||
}
|
||||
if spy.lastInput.SessionID != "sess-caller" {
|
||||
t.Errorf("SessionID = %q, want sess-caller (flag is on)", spy.lastInput.SessionID)
|
||||
}
|
||||
}
|
||||
|
||||
// insertTestTaskNoStep inserts a task row with no workflow_step_id bound
|
||||
// (the schema default is ” — see tasks.workflow_step_id NOT NULL DEFAULT
|
||||
// ”), exercising the AC-7/AC-55(1) precondition RecordAgentDecision must
|
||||
|
||||
@@ -361,6 +361,18 @@ type DashboardService struct {
|
||||
routingProvider RoutingProvider // optional; nil disables /routing endpoints (503)
|
||||
attemptLister RouteAttemptLister // optional; nil disables attempt embedding on run-detail responses
|
||||
runResolver RunResolver // optional; nil means status-change activity rows have no run_id
|
||||
// officeSessionIdentity gates RecordAgentDecision's use of the caller's
|
||||
// own session id. Defaults false (zero value); set via
|
||||
// SetOfficeSessionIdentity, wired from features.officeSessionIdentity.
|
||||
officeSessionIdentity bool
|
||||
}
|
||||
|
||||
// SetOfficeSessionIdentity wires the features.officeSessionIdentity flag.
|
||||
// When true, RecordAgentDecision forwards the decider's own calling session
|
||||
// id so RecordDecision re-evaluates against it instead of the task's
|
||||
// most-recently-started ("active") session. Defaults false.
|
||||
func (s *DashboardService) SetOfficeSessionIdentity(enabled bool) {
|
||||
s.officeSessionIdentity = enabled
|
||||
}
|
||||
|
||||
// SetRoutingProvider wires the provider-routing seam used by the
|
||||
|
||||
@@ -74,6 +74,15 @@ type RecordDecisionInput struct {
|
||||
DeciderID string
|
||||
Role string
|
||||
Comment string
|
||||
// SessionID, when non-empty, names the decider's own calling session.
|
||||
// RecordDecision re-evaluates against this session instead of the
|
||||
// task's most-recently-started ("active") session, so a reviewer/
|
||||
// approver's decision is checked against the seats their own session
|
||||
// occupies rather than whichever session happens to be newest.
|
||||
// Callers only populate this behind features.officeSessionIdentity
|
||||
// (see office/dashboard.DashboardService.SetOfficeSessionIdentity);
|
||||
// leaving it empty preserves the existing resolveActiveSessionID path.
|
||||
SessionID string
|
||||
}
|
||||
|
||||
// RecordDecisionResult mirrors engine.RecordDecisionResult plus the
|
||||
@@ -270,9 +279,15 @@ func (d *Dispatcher) RecordDecision(ctx context.Context, in RecordDecisionInput)
|
||||
if in.TaskID == "" {
|
||||
return RecordDecisionResult{}, fmt.Errorf("task_id is required")
|
||||
}
|
||||
sessionID, err := d.resolveActiveSessionID(ctx, in.TaskID)
|
||||
var sessionID string
|
||||
var err error
|
||||
if in.SessionID != "" {
|
||||
sessionID, err = d.resolveDeciderSessionID(ctx, in.TaskID, in.SessionID)
|
||||
} else {
|
||||
sessionID, err = d.resolveActiveSessionID(ctx, in.TaskID)
|
||||
}
|
||||
if err != nil {
|
||||
return RecordDecisionResult{}, fmt.Errorf("resolve active session: %w", err)
|
||||
return RecordDecisionResult{}, fmt.Errorf("resolve decider session: %w", err)
|
||||
}
|
||||
result, err := d.engine.RecordParticipantDecision(ctx, sessionID, engine.DecisionInfo{
|
||||
TaskID: in.TaskID,
|
||||
@@ -368,9 +383,66 @@ func (d *Dispatcher) resolveActiveSessionID(ctx context.Context, taskID string)
|
||||
return session.ID, nil
|
||||
}
|
||||
|
||||
// resolveDeciderSessionID validates the decider's own calling session
|
||||
// (RecordDecisionInput.SessionID) rather than resolving the task's active
|
||||
// session. It returns "" (session_unresolvable, not an error) when the
|
||||
// session is missing, belongs to a different task, or is not in one of the
|
||||
// active states — mirroring resolveActiveSessionID's own never-an-error
|
||||
// contract for "no session resolvable" (AC-16a) — so a stale or foreign
|
||||
// session id degrades to the same "skip re-evaluation" behavior an
|
||||
// unresolvable active session already gets, rather than rejecting the
|
||||
// decision outright.
|
||||
//
|
||||
// The active-state set mirrors GetActiveTaskSessionByTaskID's own query
|
||||
// (internal/task/repository/sqlite/session.go) — there is no shared
|
||||
// models-level predicate for that set, so this local copy must be kept in
|
||||
// sync with it.
|
||||
func (d *Dispatcher) resolveDeciderSessionID(ctx context.Context, taskID, sessionID string) (string, error) {
|
||||
session, err := d.sessions.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
if errors.Is(err, taskmodels.ErrTaskSessionNotFound) {
|
||||
d.logSessionUnresolvable(taskID, sessionID, "not_found")
|
||||
return "", nil
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
if session == nil || session.TaskID != taskID {
|
||||
d.logSessionUnresolvable(taskID, sessionID, "foreign")
|
||||
return "", nil
|
||||
}
|
||||
if !isActiveSessionState(session.State) {
|
||||
d.logSessionUnresolvable(taskID, sessionID, "terminal")
|
||||
return "", nil
|
||||
}
|
||||
return session.ID, nil
|
||||
}
|
||||
|
||||
func (d *Dispatcher) logSessionUnresolvable(taskID, sessionID, reason string) {
|
||||
d.logger.Debug("resolveDeciderSessionID: session unresolvable, falling back to blank",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("supplied_session_id", sessionID),
|
||||
zap.String("reason", reason))
|
||||
}
|
||||
|
||||
func isActiveSessionState(state taskmodels.TaskSessionState) bool {
|
||||
switch state {
|
||||
case taskmodels.TaskSessionStateCreated,
|
||||
taskmodels.TaskSessionStateStarting,
|
||||
taskmodels.TaskSessionStateRunning,
|
||||
taskmodels.TaskSessionStateWaitingForInput:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// resolveLatestSessionID returns the F38 "any session" id (the task's
|
||||
// most recent session regardless of state), or "" when the task has never
|
||||
// had one.
|
||||
// had one. Like resolveActiveSessionID and resolveDeciderSessionID, this is
|
||||
// scoped to the given taskID — GetTaskSessionByTaskID's own query already
|
||||
// filters by task, so unlike resolveDeciderSessionID (which validates a
|
||||
// caller-supplied session id and must check task ownership itself), there
|
||||
// is no cross-task session id to smuggle in here.
|
||||
func (d *Dispatcher) resolveLatestSessionID(ctx context.Context, taskID string) (string, error) {
|
||||
session, err := d.sessions.GetTaskSessionByTaskID(ctx, taskID)
|
||||
if err != nil {
|
||||
|
||||
@@ -302,3 +302,129 @@ func TestDispatcher_EvaluateStepQuorum_PropagatesEngineError(t *testing.T) {
|
||||
t.Fatalf("err = %v, want wrapped engine error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatcher_RecordDecision_UsesSuppliedSessionOverActiveSession is the
|
||||
// office-session-identity happy path: when RecordDecisionInput.SessionID is
|
||||
// populated, the dispatcher resolves and forwards THAT session rather than
|
||||
// the task's most-recently-started ("active") session — even when the two
|
||||
// differ, proving the decider's own session wins.
|
||||
func TestDispatcher_RecordDecision_UsesSuppliedSessionOverActiveSession(t *testing.T) {
|
||||
eng := &fakeEngine{decisionResult: engine.RecordDecisionResult{DecisionID: "decision-1"}}
|
||||
sessions := &fakeSessions{
|
||||
// The "active" (most-recently-started) session is a different one
|
||||
// than the decider's own — proves SessionID, not activeSession, wins.
|
||||
activeSession: &taskmodels.TaskSession{ID: "sess-runner"},
|
||||
byID: map[string]*taskmodels.TaskSession{
|
||||
"sess-reviewer": {ID: "sess-reviewer", TaskID: "task-1", State: taskmodels.TaskSessionStateRunning},
|
||||
},
|
||||
}
|
||||
d := New(eng, sessions, logger.Default())
|
||||
|
||||
_, err := d.RecordDecision(context.Background(), RecordDecisionInput{
|
||||
TaskID: "task-1", StepID: "review", ParticipantID: "participant-1",
|
||||
Decision: "approved", SessionID: "sess-reviewer",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordDecision: %v", err)
|
||||
}
|
||||
if eng.decisionSession != "sess-reviewer" {
|
||||
t.Errorf("session id = %q, want sess-reviewer (decider's own session, not the active one)", eng.decisionSession)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatcher_RecordDecision_ForeignSessionFallsBackToUnresolvable
|
||||
// covers the "session_unresolvable, not an error" contract: a supplied
|
||||
// session that belongs to a different task must not leak cross-task state
|
||||
// into re-evaluation, but must also not fail the decision outright — it
|
||||
// degrades to the same blank-session-id "skip re-evaluation" behavior
|
||||
// AC-16a already gives an unresolvable active session.
|
||||
func TestDispatcher_RecordDecision_ForeignSessionFallsBackToUnresolvable(t *testing.T) {
|
||||
eng := &fakeEngine{decisionResult: engine.RecordDecisionResult{DecisionID: "decision-1"}}
|
||||
sessions := &fakeSessions{
|
||||
byID: map[string]*taskmodels.TaskSession{
|
||||
"sess-foreign": {ID: "sess-foreign", TaskID: "other-task", State: taskmodels.TaskSessionStateRunning},
|
||||
},
|
||||
}
|
||||
d := New(eng, sessions, logger.Default())
|
||||
|
||||
_, err := d.RecordDecision(context.Background(), RecordDecisionInput{
|
||||
TaskID: "task-1", StepID: "review", ParticipantID: "participant-1",
|
||||
Decision: "approved", SessionID: "sess-foreign",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordDecision: %v, want success (session_unresolvable, not an error)", err)
|
||||
}
|
||||
if !eng.decisionCalled {
|
||||
t.Fatal("engine RecordParticipantDecision not invoked")
|
||||
}
|
||||
if eng.decisionSession != "" {
|
||||
t.Errorf("session id = %q, want blank: foreign session must not be forwarded", eng.decisionSession)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatcher_RecordDecision_TerminalSessionFallsBackToUnresolvable
|
||||
// covers the other session_unresolvable case: a supplied session that
|
||||
// belongs to the right task but is in a terminal (non-active) state.
|
||||
func TestDispatcher_RecordDecision_TerminalSessionFallsBackToUnresolvable(t *testing.T) {
|
||||
eng := &fakeEngine{decisionResult: engine.RecordDecisionResult{DecisionID: "decision-1"}}
|
||||
sessions := &fakeSessions{
|
||||
byID: map[string]*taskmodels.TaskSession{
|
||||
"sess-done": {ID: "sess-done", TaskID: "task-1", State: taskmodels.TaskSessionStateCompleted},
|
||||
},
|
||||
}
|
||||
d := New(eng, sessions, logger.Default())
|
||||
|
||||
_, err := d.RecordDecision(context.Background(), RecordDecisionInput{
|
||||
TaskID: "task-1", StepID: "review", ParticipantID: "participant-1",
|
||||
Decision: "approved", SessionID: "sess-done",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordDecision: %v, want success (session_unresolvable, not an error)", err)
|
||||
}
|
||||
if eng.decisionSession != "" {
|
||||
t.Errorf("session id = %q, want blank: terminal session must not be forwarded", eng.decisionSession)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatcher_RecordDecision_MissingSessionFallsBackToUnresolvable
|
||||
// covers the third session_unresolvable case: a supplied session id that
|
||||
// does not exist at all (e.g. a stale/deleted session).
|
||||
func TestDispatcher_RecordDecision_MissingSessionFallsBackToUnresolvable(t *testing.T) {
|
||||
eng := &fakeEngine{decisionResult: engine.RecordDecisionResult{DecisionID: "decision-1"}}
|
||||
sessions := &fakeSessions{} // byID is nil: GetTaskSession returns ErrTaskSessionNotFound
|
||||
d := New(eng, sessions, logger.Default())
|
||||
|
||||
_, err := d.RecordDecision(context.Background(), RecordDecisionInput{
|
||||
TaskID: "task-1", StepID: "review", ParticipantID: "participant-1",
|
||||
Decision: "approved", SessionID: "sess-missing",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordDecision: %v, want success (session_unresolvable, not an error)", err)
|
||||
}
|
||||
if eng.decisionSession != "" {
|
||||
t.Errorf("session id = %q, want blank: missing session must not be forwarded", eng.decisionSession)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDispatcher_RecordDecision_SessionLookupErrorPropagates proves a
|
||||
// genuine session-store failure while resolving the supplied SessionID is
|
||||
// not silently swallowed as session_unresolvable — mirroring
|
||||
// TestDispatcher_RecordDecision_PropagatesActiveSessionLookupError for the
|
||||
// resolveActiveSessionID path.
|
||||
func TestDispatcher_RecordDecision_SessionLookupErrorPropagates(t *testing.T) {
|
||||
dbErr := errors.New("db down")
|
||||
eng := &fakeEngine{}
|
||||
sessions := &fakeSessions{byIDErr: dbErr}
|
||||
d := New(eng, sessions, logger.Default())
|
||||
|
||||
_, err := d.RecordDecision(context.Background(), RecordDecisionInput{
|
||||
TaskID: "task-1", StepID: "review", ParticipantID: "participant-1",
|
||||
Decision: "approved", SessionID: "sess-1",
|
||||
})
|
||||
if !errors.Is(err, dbErr) {
|
||||
t.Fatalf("err = %v, want wrapped db error", err)
|
||||
}
|
||||
if eng.decisionCalled {
|
||||
t.Error("engine should not be called when session lookup fails")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,7 +97,7 @@ func TestPrepareSessionForStart_PropagatesInheritedEnvironment(t *testing.T) {
|
||||
ParentID: "parent",
|
||||
Metadata: map[string]interface{}{"workspace": map[string]interface{}{"mode": "inherit_parent"}},
|
||||
}
|
||||
sessionID, _, err := svc.prepareSessionForStart(ctx, childTask, "profile-1", "exec-1", "", "")
|
||||
sessionID, _, err := svc.prepareSessionForStart(ctx, childTask, "profile-1", "profile-1", "exec-1", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("prepareSessionForStart: %v", err)
|
||||
}
|
||||
@@ -125,7 +125,7 @@ func TestPrepareSessionForStart_InheritedWorkspaceMissingCompensatesSession(t *t
|
||||
ID: "child-missing",
|
||||
ParentID: "parent-missing",
|
||||
Metadata: map[string]interface{}{"workspace": map[string]interface{}{"mode": "inherit_parent"}},
|
||||
}, "profile-1", "exec-1", "", "")
|
||||
}, "profile-1", "profile-1", "exec-1", "", "")
|
||||
if !errors.Is(err, models.ErrWorkspaceReuseUnsafe) {
|
||||
t.Fatalf("prepareSessionForStart error = %v, want workspace reuse unsafe", err)
|
||||
}
|
||||
|
||||
@@ -63,6 +63,12 @@ type ServiceConfig struct {
|
||||
// turn for an agent that advertised prompt queueing. Independent of
|
||||
// ClaudeBackgroundPromptHandoff, which covers the foreground-idle handoff.
|
||||
ClaudeMidTurnSteering bool
|
||||
|
||||
// OfficeSessionIdentity keys an Office task's session identity on the
|
||||
// run's own agent instead of the task's runner seat. Off by default;
|
||||
// enabling it exposes pre-existing duplicate (task_id, agent_profile_id)
|
||||
// rows until the companion unique-index fix has shipped.
|
||||
OfficeSessionIdentity bool
|
||||
}
|
||||
|
||||
// AttachmentReader is the narrow attachment-store seam needed when the
|
||||
|
||||
@@ -151,7 +151,7 @@ func (s *Service) EnsureSession(ctx context.Context, taskID string, opts ...Ensu
|
||||
// singleton human user). For kanban tasks, it falls back to the existing
|
||||
// is_primary-first lookup. Returns nil when no matching session exists.
|
||||
func (s *Service) findExistingSession(ctx context.Context, taskID string) *EnsureSessionResponse {
|
||||
if office := s.findOfficeSessionForResume(ctx, taskID); office != nil {
|
||||
if office, isOffice := s.findOfficeSessionForResume(ctx, taskID); isOffice {
|
||||
return office
|
||||
}
|
||||
sessions, err := s.repo.ListTaskSessions(ctx, taskID)
|
||||
@@ -168,24 +168,36 @@ func (s *Service) findExistingSession(ctx context.Context, taskID string) *Ensur
|
||||
}
|
||||
|
||||
// findOfficeSessionForResume implements the office-only branch of advanced-mode
|
||||
// resume. Returns nil when the task isn't office, when no per-agent session
|
||||
// has been created yet, or when no relevant agent identity is available.
|
||||
// (The "create on demand" branch is handled by EnsureSession's fallthrough,
|
||||
// not here — keeping findExistingSession a pure lookup.)
|
||||
func (s *Service) findOfficeSessionForResume(ctx context.Context, taskID string) *EnsureSessionResponse {
|
||||
// resume. The second return value reports whether the task is Office-owned.
|
||||
// Returns nil when no per-agent session has been created yet or when no
|
||||
// relevant agent identity is available. An Office task must not fall through to
|
||||
// the generic primary/newest lookup, because that lookup can return another
|
||||
// participant's session. When the task has no projected runner (for example,
|
||||
// a task created with only metadata.agent_profile_id), resolve the same profile
|
||||
// EnsureSession would use for creation and look up that exact session. The
|
||||
// "create on demand" branch is handled by EnsureSession after this pure lookup.
|
||||
func (s *Service) findOfficeSessionForResume(ctx context.Context, taskID string) (*EnsureSessionResponse, bool) {
|
||||
task, err := s.repo.GetTask(ctx, taskID)
|
||||
if err != nil || task == nil || !task.IsFromOffice {
|
||||
return nil
|
||||
return nil, false
|
||||
}
|
||||
agentID := s.agentForViewer(ctx, task)
|
||||
if agentID == "" {
|
||||
return nil
|
||||
// A task can be Office-owned without a projected runner. This is
|
||||
// common for API-created review tasks whose concrete profile lives in
|
||||
// metadata. Match the profile resolution used by EnsureSession so an
|
||||
// already-created run session is reused while still avoiding a
|
||||
// participant-agnostic newest-session fallback.
|
||||
agentID, _ = s.resolveTaskAgentProfile(ctx, task)
|
||||
}
|
||||
if agentID == "" {
|
||||
return nil, true
|
||||
}
|
||||
sess, err := s.repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agentID)
|
||||
if err != nil || sess == nil {
|
||||
return nil
|
||||
return nil, true
|
||||
}
|
||||
return s.existingResponse(ctx, taskID, sess, "existing_office_agent")
|
||||
return s.existingResponse(ctx, taskID, sess, "existing_office_agent"), true
|
||||
}
|
||||
|
||||
// agentForViewer resolves the agent_profile_id whose session should be
|
||||
|
||||
@@ -79,6 +79,86 @@ func TestFindExistingSession_OfficeTaskWithViewerAgent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindExistingSession_OfficeTaskWithOnlyOtherAgentSessionReturnsNil(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedOfficeTaskAndSessions(t, repo)
|
||||
if err := repo.DeleteTaskSession(context.Background(), "s-assignee"); err != nil {
|
||||
t.Fatalf("delete assignee session: %v", err)
|
||||
}
|
||||
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
|
||||
|
||||
if got := svc.findExistingSession(context.Background(), "t-office"); got != nil {
|
||||
t.Fatalf("expected no assignee session, got %q", got.SessionID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindExistingSession_OfficeTaskUsesResolvedMetadataProfile(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
seedOfficeTaskAndSessions(t, repo)
|
||||
if err := repo.DeleteTaskSession(ctx, "s-assignee"); err != nil {
|
||||
t.Fatalf("delete assignee session: %v", err)
|
||||
}
|
||||
task, err := repo.GetTask(ctx, "t-office")
|
||||
if err != nil {
|
||||
t.Fatalf("get task: %v", err)
|
||||
}
|
||||
task.AssigneeAgentProfileID = ""
|
||||
task.Metadata = map[string]interface{}{"agent_profile_id": "agent-reviewer"}
|
||||
if err := repo.UpdateTask(ctx, task); err != nil {
|
||||
t.Fatalf("update task: %v", err)
|
||||
}
|
||||
|
||||
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
|
||||
resp := svc.findExistingSession(ctx, "t-office")
|
||||
if resp == nil {
|
||||
t.Fatal("expected metadata-resolved session, got nil")
|
||||
}
|
||||
if resp.SessionID != "s-reviewer" {
|
||||
t.Fatalf("session_id = %q, want s-reviewer", resp.SessionID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareTaskSession_OfficeFlagUsesAssigneeForSessionIdentity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
seedOfficeTaskAndSessions(t, repo)
|
||||
if err := repo.DeleteTaskSession(ctx, "s-assignee"); err != nil {
|
||||
t.Fatalf("delete assignee session: %v", err)
|
||||
}
|
||||
if err := repo.DeleteTaskSession(ctx, "s-reviewer"); err != nil {
|
||||
t.Fatalf("delete reviewer session: %v", err)
|
||||
}
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks["t-office"] = &v1.Task{
|
||||
ID: "t-office",
|
||||
WorkspaceID: "ws-r",
|
||||
WorkflowID: "wf-r",
|
||||
Title: "Office",
|
||||
State: v1.TaskStateInProgress,
|
||||
}
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
|
||||
svc.config.OfficeSessionIdentity = true
|
||||
|
||||
sessionID, err := svc.PrepareTaskSession(
|
||||
ctx, "t-office", "execution-profile", "", "", "", false,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("PrepareTaskSession: %v", err)
|
||||
}
|
||||
session, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("get prepared session: %v", err)
|
||||
}
|
||||
if session.AgentProfileID != "agent-assignee" {
|
||||
t.Fatalf("session identity = %q, want agent-assignee", session.AgentProfileID)
|
||||
}
|
||||
if session.ExecutionProfileID != "execution-profile" {
|
||||
t.Fatalf("execution profile = %q, want execution-profile", session.ExecutionProfileID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFindExistingSession_KanbanTask_UsesPrimary keeps the kanban-side
|
||||
// is_primary lookup intact under the new gating.
|
||||
func TestFindExistingSession_KanbanTask_UsesPrimary(t *testing.T) {
|
||||
|
||||
@@ -126,9 +126,12 @@ func TestFindOfficeSessionForResumeUsesCanonicalOfficeProjection(t *testing.T) {
|
||||
if tt.viewer != "" {
|
||||
ctx = WithViewerAgent(ctx, tt.viewer)
|
||||
}
|
||||
got := svc.findOfficeSessionForResume(ctx, "task1")
|
||||
if (got != nil) != tt.wantOffice {
|
||||
t.Fatalf("office session found = %v, want %v", got != nil, tt.wantOffice)
|
||||
got, isOffice := svc.findOfficeSessionForResume(ctx, "task1")
|
||||
if isOffice != tt.wantOffice {
|
||||
t.Fatalf("is office task = %v, want %v", isOffice, tt.wantOffice)
|
||||
}
|
||||
if (got != nil) != (tt.wantOffice && tt.viewer != "") {
|
||||
t.Fatalf("office session found = %v, want %v", got != nil, tt.wantOffice && tt.viewer != "")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -322,7 +322,10 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr
|
||||
// EnsureSessionForAgent so runs + advanced-mode reuse one row.
|
||||
// prepareSessionForStart also propagates any inherited workspace
|
||||
// environment (inherit_parent / shared_group) onto the new session.
|
||||
sessionID, sessionCreated, err := s.prepareSessionForStart(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
// A manual prepare call has no scheduler-owned Office run identity. Keep the
|
||||
// participant slot empty so createStartSession uses the task assignee, while
|
||||
// agentProfileID remains the concrete execution profile.
|
||||
sessionID, sessionCreated, err := s.prepareSessionForStart(ctx, task, agentProfileID, "", executorID, executorProfileID, workflowStepID)
|
||||
if err != nil {
|
||||
s.logger.Error("failed to prepare session",
|
||||
zap.String("task_id", taskID),
|
||||
@@ -1051,7 +1054,7 @@ func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID s
|
||||
// Prepare session first so we have the sessionID for config context injection.
|
||||
// For office tasks, replace the per-launch PrepareSession with the per-(task,
|
||||
// agent) EnsureSessionForAgent so runs reuse one row across turns.
|
||||
sessionID, sessionCreated, err := s.prepareSessionForStart(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
sessionID, sessionCreated, err := s.prepareSessionForStart(ctx, task, agentProfileID, officeAgentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1394,9 +1397,9 @@ func validateOfficeLaunchEnv(taskID string, env map[string]string) error {
|
||||
// propagateInheritedEnvironment is a no-op for tasks without a workspace policy.
|
||||
func (s *Service) prepareSessionForStart(
|
||||
ctx context.Context, task *v1.Task,
|
||||
agentProfileID, executorID, executorProfileID, workflowStepID string,
|
||||
agentProfileID, officeAgentProfileID, executorID, executorProfileID, workflowStepID string,
|
||||
) (string, bool, error) {
|
||||
sessionID, created, err := s.createStartSession(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
sessionID, created, err := s.createStartSession(ctx, task, agentProfileID, officeAgentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
@@ -1585,25 +1588,60 @@ func (s *Service) recordDynamicRouteResolutionFailure(
|
||||
}
|
||||
|
||||
// createStartSession picks the right session-creation path for the task:
|
||||
// Office tasks with an assignee use the per-(task, agent) EnsureSessionForAgent
|
||||
// (so runs reuse one row across turns); kanban / quick-chat fall through to
|
||||
// the per-launch PrepareSession used since day one.
|
||||
// Office tasks use the per-(task, agent) EnsureSessionForAgent path (so runs
|
||||
// reuse one row across turns); kanban / quick-chat fall through to the
|
||||
// per-launch PrepareSession used since day one. An Office task can be created
|
||||
// before its runner seat is projected, so the run's captured identity or its
|
||||
// resolved execution profile supplies the owner until the seat is assigned.
|
||||
//
|
||||
// The session-owner identity passed to EnsureSessionForAgentWithCreation is,
|
||||
// by default, the task's runner seat (dbTask.AssigneeAgentProfileID) — even
|
||||
// for a reviewer/approver run whose agent differs from the runner, which
|
||||
// wrongly binds that run's session (and later its decisions) to the runner's
|
||||
// identity. When features.officeSessionIdentity is on, officeAgentProfileID
|
||||
// (the run's own agent, captured by the caller before step/routing overrides
|
||||
// mutate agentProfileID) is used instead so each participant agent gets its
|
||||
// own session per task.
|
||||
func (s *Service) createStartSession(
|
||||
ctx context.Context, task *v1.Task,
|
||||
agentProfileID, executorID, executorProfileID, workflowStepID string,
|
||||
agentProfileID, officeAgentProfileID, executorID, executorProfileID, workflowStepID string,
|
||||
) (string, bool, error) {
|
||||
dbTask, err := s.repo.GetTask(ctx, task.ID)
|
||||
if err == nil && dbTask != nil && dbTask.IsFromOffice && dbTask.AssigneeAgentProfileID != "" {
|
||||
session, created, ensureErr := s.executor.EnsureSessionForAgentWithCreation(
|
||||
ctx, task, dbTask.AssigneeAgentProfileID, agentProfileID, executorID, executorProfileID,
|
||||
)
|
||||
if ensureErr != nil {
|
||||
return "", false, ensureErr
|
||||
}
|
||||
return session.ID, created, nil
|
||||
if err != nil || dbTask == nil || !dbTask.IsFromOffice {
|
||||
sessionID, prepareErr := s.executor.PrepareSession(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
return sessionID, prepareErr == nil, prepareErr
|
||||
}
|
||||
sessionID, err := s.executor.PrepareSession(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
return sessionID, err == nil, err
|
||||
|
||||
sessionOwnerID := s.officeSessionOwnerID(dbTask, agentProfileID, officeAgentProfileID)
|
||||
if sessionOwnerID == "" {
|
||||
sessionID, prepareErr := s.executor.PrepareSession(ctx, task, agentProfileID, executorID, executorProfileID, workflowStepID)
|
||||
return sessionID, prepareErr == nil, prepareErr
|
||||
}
|
||||
session, created, ensureErr := s.executor.EnsureSessionForAgentWithCreation(
|
||||
ctx, task, sessionOwnerID, agentProfileID, executorID, executorProfileID,
|
||||
)
|
||||
if ensureErr != nil {
|
||||
return "", false, ensureErr
|
||||
}
|
||||
return session.ID, created, nil
|
||||
}
|
||||
|
||||
// officeSessionOwnerID selects the stable identity for an Office session. A
|
||||
// projected runner wins by default. When identity separation is enabled, a
|
||||
// scheduler-captured participant identity wins for assigned runs. Before a
|
||||
// runner seat is projected, the captured identity or execution profile keeps
|
||||
// the session on the Office path instead of creating a generic workspace row.
|
||||
func (s *Service) officeSessionOwnerID(task *models.Task, agentProfileID, officeAgentProfileID string) string {
|
||||
if task.AssigneeAgentProfileID == "" {
|
||||
if officeAgentProfileID != "" {
|
||||
return officeAgentProfileID
|
||||
}
|
||||
return agentProfileID
|
||||
}
|
||||
if s.config.OfficeSessionIdentity && officeAgentProfileID != "" {
|
||||
return officeAgentProfileID
|
||||
}
|
||||
return task.AssigneeAgentProfileID
|
||||
}
|
||||
|
||||
// moveTaskToWorkflowStep moves a task to the target workflow step if provided and different from current.
|
||||
|
||||
@@ -173,7 +173,7 @@ func TestCreateStartSession_KanbanRunnerCreatesDistinctSession(t *testing.T) {
|
||||
if isOffice {
|
||||
t.Fatal("kanban task with a runner was classified as office-owned")
|
||||
}
|
||||
sessionID, _, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "", "", "")
|
||||
sessionID, _, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "copilot-runner", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("create start session: %v", err)
|
||||
}
|
||||
@@ -226,7 +226,7 @@ func TestCreateStartSession_OfficeRunnerReusesPersistentSession(t *testing.T) {
|
||||
if !isOffice {
|
||||
t.Fatal("office-owned assigned task was not classified as office")
|
||||
}
|
||||
sessionID, created, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "", "", "")
|
||||
sessionID, created, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "copilot-runner", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("create start session: %v", err)
|
||||
}
|
||||
@@ -238,6 +238,177 @@ func TestCreateStartSession_OfficeRunnerReusesPersistentSession(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateStartSession_OfficeUnassignedReusesResolvedProfileSession(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-office-unassigned", Name: "Office", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-office-unassigned", WorkspaceID: "ws-office-unassigned", Name: "Office", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
if err := seedWorkflowStep(t, repo, "step-office-unassigned"); err != nil {
|
||||
t.Fatalf("create workflow step: %v", err)
|
||||
}
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-office-unassigned", WorkspaceID: "ws-office-unassigned", WorkflowID: "wf-office-unassigned", WorkflowStepID: "step-office-unassigned",
|
||||
Title: "Office task", State: v1.TaskStateInProgress, ProjectID: "office-project",
|
||||
Metadata: map[string]interface{}{models.MetaKeyAgentProfileID: "ceo-reviewer"},
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
|
||||
task, err := repo.GetTask(ctx, "task-office-unassigned")
|
||||
if err != nil {
|
||||
t.Fatalf("get task: %v", err)
|
||||
}
|
||||
if !task.IsFromOffice {
|
||||
t.Fatal("office task was not projected as office-owned")
|
||||
}
|
||||
if task.AssigneeAgentProfileID != "" {
|
||||
t.Fatalf("task unexpectedly has an assignee: %q", task.AssigneeAgentProfileID)
|
||||
}
|
||||
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{})
|
||||
firstID, firstCreated, err := svc.createStartSession(ctx, task.ToAPI(), "ceo-reviewer", "", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("first create start session: %v", err)
|
||||
}
|
||||
if !firstCreated {
|
||||
t.Fatal("first office launch should create a session")
|
||||
}
|
||||
|
||||
secondID, secondCreated, err := svc.createStartSession(ctx, task.ToAPI(), "ceo-reviewer", "", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("second create start session: %v", err)
|
||||
}
|
||||
if secondCreated {
|
||||
t.Fatal("second office launch should reuse the profile session")
|
||||
}
|
||||
if secondID != firstID {
|
||||
t.Fatalf("reused session = %q, want first session %q", secondID, firstID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCreateStartSession_ReviewerRunFlagOff is the regression baseline: a
|
||||
// review run whose agent (ceo-reviewer) differs from the task's runner seat
|
||||
// (pm-runner) lands in the runner's session when features.officeSessionIdentity
|
||||
// is off (the default), reproducing the FORBIDDEN bug this task fixes —
|
||||
// record_step_decision_kandev resolves the session's AgentProfileID as the
|
||||
// decider identity, so the reviewer's own decision is checked against seats
|
||||
// the runner (not the reviewer) occupies.
|
||||
func TestCreateStartSession_ReviewerRunFlagOff(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-review", Name: "Review", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-review", WorkspaceID: "ws-review", Name: "Review", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
if err := seedWorkflowStep(t, repo, "step-review-start"); err != nil {
|
||||
t.Fatalf("create workflow step: %v", err)
|
||||
}
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-review", WorkspaceID: "ws-review", WorkflowID: "wf-review", WorkflowStepID: "step-review-start",
|
||||
Title: "Review task", State: v1.TaskStateInProgress, ProjectID: "office-project", AssigneeAgentProfileID: "pm-runner",
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
if err := repo.CreateTaskSession(ctx, &models.TaskSession{
|
||||
ID: "runner-session", TaskID: "task-review", AgentProfileID: "pm-runner",
|
||||
State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create runner session: %v", err)
|
||||
}
|
||||
|
||||
task, err := repo.GetTask(ctx, "task-review")
|
||||
if err != nil {
|
||||
t.Fatalf("get task: %v", err)
|
||||
}
|
||||
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{})
|
||||
|
||||
// A reviewer run: the run's own agent (ceo-reviewer) differs from the
|
||||
// task's runner seat (pm-runner).
|
||||
sessionID, _, err := svc.createStartSession(ctx, task.ToAPI(), "ceo-reviewer", "ceo-reviewer", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("create start session: %v", err)
|
||||
}
|
||||
session, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
if session.AgentProfileID != "pm-runner" {
|
||||
t.Fatalf("session owner = %q, want pm-runner (flag off preserves runner-keyed identity)", session.AgentProfileID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCreateStartSession_ReviewerRunFlagOnGetsOwnSession is the fix's green
|
||||
// case: with features.officeSessionIdentity on, the same reviewer run lands
|
||||
// in a session keyed on the run's own agent (ceo-reviewer), not the task's
|
||||
// runner seat (pm-runner).
|
||||
func TestCreateStartSession_ReviewerRunFlagOnGetsOwnSession(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-review2", Name: "Review2", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-review2", WorkspaceID: "ws-review2", Name: "Review2", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
if err := seedWorkflowStep(t, repo, "step-review2-start"); err != nil {
|
||||
t.Fatalf("create workflow step: %v", err)
|
||||
}
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-review2", WorkspaceID: "ws-review2", WorkflowID: "wf-review2", WorkflowStepID: "step-review2-start",
|
||||
Title: "Review task", State: v1.TaskStateInProgress, ProjectID: "office-project", AssigneeAgentProfileID: "pm-runner",
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
if err := repo.CreateTaskSession(ctx, &models.TaskSession{
|
||||
ID: "runner-session2", TaskID: "task-review2", AgentProfileID: "pm-runner",
|
||||
State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create runner session: %v", err)
|
||||
}
|
||||
|
||||
task, err := repo.GetTask(ctx, "task-review2")
|
||||
if err != nil {
|
||||
t.Fatalf("get task: %v", err)
|
||||
}
|
||||
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{})
|
||||
svc.config.OfficeSessionIdentity = true
|
||||
|
||||
sessionID, created, err := svc.createStartSession(ctx, task.ToAPI(), "ceo-reviewer", "ceo-reviewer", "", "", "")
|
||||
if err != nil {
|
||||
t.Fatalf("create start session: %v", err)
|
||||
}
|
||||
if !created {
|
||||
t.Fatal("reviewer run should create its own session, not reuse the runner's")
|
||||
}
|
||||
if sessionID == "runner-session2" {
|
||||
t.Fatal("reviewer run landed in the runner's session instead of its own")
|
||||
}
|
||||
session, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
if session.AgentProfileID != "ceo-reviewer" {
|
||||
t.Fatalf("session owner = %q, want ceo-reviewer", session.AgentProfileID)
|
||||
}
|
||||
}
|
||||
|
||||
func newCoordinatorStopTestService(
|
||||
repo repoStore,
|
||||
taskRepo scheduler.TaskRepository,
|
||||
|
||||
@@ -76,6 +76,15 @@ features:
|
||||
dev: "false"
|
||||
e2e: "false"
|
||||
|
||||
# High-risk Office per-agent session identity experiment: keys an Office
|
||||
# task's session on the run's own agent instead of the runner seat. Off in
|
||||
# every profile until the companion (task_id, agent_profile_id) unique-index
|
||||
# fix has shipped, since pre-existing duplicate rows would otherwise be exposed.
|
||||
KANDEV_FEATURES_OFFICE_SESSION_IDENTITY:
|
||||
prod: "false"
|
||||
dev: "false"
|
||||
e2e: "false"
|
||||
|
||||
# ── Mock providers ────────────────────────────────────────────────
|
||||
# Swap real third-party services for in-memory fakes. Production
|
||||
# leaves all of these unset (real APIs). Dev opts into the mock agent
|
||||
|
||||
@@ -120,6 +120,23 @@ var registrations = []runtimeFlagRegistration{
|
||||
read: func(cfg *config.Config) bool { return cfg.Features.ClaudeMidTurnSteering },
|
||||
apply: func(cfg *config.Config, value bool) { cfg.Features.ClaudeMidTurnSteering = value },
|
||||
},
|
||||
{
|
||||
definition: RuntimeFlagDefinition{
|
||||
Key: "features.officeSessionIdentity",
|
||||
EnvVar: "KANDEV_FEATURES_OFFICE_SESSION_IDENTITY",
|
||||
Kind: KindFeature,
|
||||
Label: "Office per-agent session identity",
|
||||
Description: "Keys an Office task's session identity on the run's own agent instead of the task's runner seat, and binds an agent's decision re-evaluation to its own calling session.",
|
||||
Stability: StabilityExperimental,
|
||||
RiskLevel: RiskHigh,
|
||||
RiskDescription: "Changes durable Office session identity: each participant agent gets its own session per task instead of sharing the runner's, and existing session rows are not migrated. " +
|
||||
"Enable only after the companion (task_id, agent_profile_id) unique-index fix has shipped, since pre-existing duplicate rows are otherwise exposed.",
|
||||
RestartRequired: true,
|
||||
Mutable: true,
|
||||
},
|
||||
read: func(cfg *config.Config) bool { return cfg.Features.OfficeSessionIdentity },
|
||||
apply: func(cfg *config.Config, value bool) { cfg.Features.OfficeSessionIdentity = value },
|
||||
},
|
||||
{
|
||||
definition: RuntimeFlagDefinition{
|
||||
Key: "debug.devMode",
|
||||
|
||||
@@ -12,6 +12,7 @@ export const defaultFeatureFlags = {
|
||||
dynamicAgentRouting: false,
|
||||
claudeBackgroundPromptHandoff: false,
|
||||
claudeMidTurnSteering: false,
|
||||
officeSessionIdentity: false,
|
||||
} as const;
|
||||
|
||||
export type FeatureName = keyof typeof defaultFeatureFlags;
|
||||
|
||||
@@ -473,6 +473,7 @@ Copying this entire file is unnecessary and can freeze old defaults in a deploym
|
||||
| Key | Environment lock | Production default | Effect |
|
||||
|---|---|---|---|
|
||||
| `features.dynamicAgentRouting` | `KANDEV_FEATURES_DYNAMIC_AGENT_ROUTING` | off | Experimental dynamic profiles with ordered provider-error fallback. |
|
||||
| `features.officeSessionIdentity` | `KANDEV_FEATURES_OFFICE_SESSION_IDENTITY` | off | Experimental Office participant sessions. Enable only after the `(task_id, agent_profile_id)` unique index is available. |
|
||||
| `debug.devMode` | `KANDEV_DEBUG_DEV_MODE` | off | High-risk diagnostic endpoints and ACP frame logging. |
|
||||
|
||||
The `KANDEV_FEATURES_*` values have no canonical YAML keys. They are selected
|
||||
|
||||
@@ -499,6 +499,7 @@ Kandev warns when its live WebSocket connection has not recovered for three seco
|
||||
**Settings > System > Feature Toggles** currently exposes:
|
||||
|
||||
- **Office mode**: experimental, medium risk, and off in the production profile by default.
|
||||
- **Office session identity**: experimental, high risk, and off in every profile by default. Enable it only after the `(task_id, agent_profile_id)` unique index is available. It gives each Office participant a separate task conversation and requires a restart.
|
||||
- **App status bar**: stable, low risk, and off in the production profile by default. Enabling it adds the desktop/tablet bar and phone Status entry after restart; disabling it again does not stop connections, metrics collection requested by other clients, or plugins. Urgent WebSocket connectivity warnings still remain visible while the feature is off.
|
||||
- **Claude background prompt handoff**: experimental, high risk, and off in every profile by default. Enabling it lets Claude Code accept another prompt after its foreground yields while recognized async subagent, `run_in_background` shell, or Monitor work remains active. ACP lifecycle gaps can misclassify activity or overlap prompts; use it only for controlled testing.
|
||||
- **Unread divider**: a per-user setting at **Settings > General > Task Actions**. It defaults off, takes effect immediately, and controls both the Slack-style **New** divider and read-cursor updates while that user's transcript view is visible.
|
||||
|
||||
Reference in New Issue
Block a user