diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index a376a77951..9b8390c684 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -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) diff --git a/apps/backend/internal/backendapp/orchestrator.go b/apps/backend/internal/backendapp/orchestrator.go index 7ee4686ea3..887715dbb9 100644 --- a/apps/backend/internal/backendapp/orchestrator.go +++ b/apps/backend/internal/backendapp/orchestrator.go @@ -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" diff --git a/apps/backend/internal/common/config/catalog.go b/apps/backend/internal/common/config/catalog.go index f5cb5e9137..776760d5fc 100644 --- a/apps/backend/internal/common/config/catalog.go +++ b/apps/backend/internal/common/config/catalog.go @@ -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"}, diff --git a/apps/backend/internal/common/config/catalog_test.go b/apps/backend/internal/common/config/catalog_test.go index 2599933db1..a729fe4464 100644 --- a/apps/backend/internal/common/config/catalog_test.go +++ b/apps/backend/internal/common/config/catalog_test.go @@ -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"}, diff --git a/apps/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go index c44f3bdd69..2c4039c07b 100644 --- a/apps/backend/internal/common/config/config.go +++ b/apps/backend/internal/common/config/config.go @@ -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. diff --git a/apps/backend/internal/mcp/handlers/agent_decision_handlers.go b/apps/backend/internal/mcp/handlers/agent_decision_handlers.go index d68cc97516..68fb5e66b9 100644 --- a/apps/backend/internal/mcp/handlers/agent_decision_handlers.go +++ b/apps/backend/internal/mcp/handlers/agent_decision_handlers.go @@ -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) { diff --git a/apps/backend/internal/office/dashboard/agent_decisions.go b/apps/backend/internal/office/dashboard/agent_decisions.go index 5b2a00659c..3108360b7a 100644 --- a/apps/backend/internal/office/dashboard/agent_decisions.go +++ b/apps/backend/internal/office/dashboard/agent_decisions.go @@ -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) } diff --git a/apps/backend/internal/office/dashboard/agent_decisions_test.go b/apps/backend/internal/office/dashboard/agent_decisions_test.go index dcf4acec6d..f91a0a3ebf 100644 --- a/apps/backend/internal/office/dashboard/agent_decisions_test.go +++ b/apps/backend/internal/office/dashboard/agent_decisions_test.go @@ -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 diff --git a/apps/backend/internal/office/dashboard/service.go b/apps/backend/internal/office/dashboard/service.go index 39a8d655ac..ce113acc6d 100644 --- a/apps/backend/internal/office/dashboard/service.go +++ b/apps/backend/internal/office/dashboard/service.go @@ -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 diff --git a/apps/backend/internal/office/engine_dispatcher/dispatcher.go b/apps/backend/internal/office/engine_dispatcher/dispatcher.go index 878f919d85..426bf8494f 100644 --- a/apps/backend/internal/office/engine_dispatcher/dispatcher.go +++ b/apps/backend/internal/office/engine_dispatcher/dispatcher.go @@ -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 { diff --git a/apps/backend/internal/office/engine_dispatcher/dispatcher_decisions_test.go b/apps/backend/internal/office/engine_dispatcher/dispatcher_decisions_test.go index a4ae419795..3b500cfb6b 100644 --- a/apps/backend/internal/office/engine_dispatcher/dispatcher_decisions_test.go +++ b/apps/backend/internal/office/engine_dispatcher/dispatcher_decisions_test.go @@ -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") + } +} diff --git a/apps/backend/internal/orchestrator/handoff_inheritance_test.go b/apps/backend/internal/orchestrator/handoff_inheritance_test.go index f13ccf8b3c..9322840747 100644 --- a/apps/backend/internal/orchestrator/handoff_inheritance_test.go +++ b/apps/backend/internal/orchestrator/handoff_inheritance_test.go @@ -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) } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index d3b05fdeda..efa712c0c0 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -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 diff --git a/apps/backend/internal/orchestrator/session_ensure.go b/apps/backend/internal/orchestrator/session_ensure.go index df503b8a2b..ae6b9b79c3 100644 --- a/apps/backend/internal/orchestrator/session_ensure.go +++ b/apps/backend/internal/orchestrator/session_ensure.go @@ -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 diff --git a/apps/backend/internal/orchestrator/session_ensure_office_test.go b/apps/backend/internal/orchestrator/session_ensure_office_test.go index 68bde00025..022ff7a490 100644 --- a/apps/backend/internal/orchestrator/session_ensure_office_test.go +++ b/apps/backend/internal/orchestrator/session_ensure_office_test.go @@ -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) { diff --git a/apps/backend/internal/orchestrator/session_ensure_test.go b/apps/backend/internal/orchestrator/session_ensure_test.go index 1c865733a7..5a4360c937 100644 --- a/apps/backend/internal/orchestrator/session_ensure_test.go +++ b/apps/backend/internal/orchestrator/session_ensure_test.go @@ -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 != "") } }) } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 86f39f8c22..09e50f1ed2 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -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. diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index ae4ba623c1..1f742f75b8 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -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, diff --git a/apps/backend/internal/profiles/profiles.yaml b/apps/backend/internal/profiles/profiles.yaml index e97e4fddf9..d5e1ddadba 100644 --- a/apps/backend/internal/profiles/profiles.yaml +++ b/apps/backend/internal/profiles/profiles.yaml @@ -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 diff --git a/apps/backend/internal/runtimeflags/registry.go b/apps/backend/internal/runtimeflags/registry.go index 8449de478f..267f6f4d62 100644 --- a/apps/backend/internal/runtimeflags/registry.go +++ b/apps/backend/internal/runtimeflags/registry.go @@ -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", diff --git a/apps/web/lib/state/slices/features/types.ts b/apps/web/lib/state/slices/features/types.ts index 9452d0c277..597abd67e9 100644 --- a/apps/web/lib/state/slices/features/types.ts +++ b/apps/web/lib/state/slices/features/types.ts @@ -12,6 +12,7 @@ export const defaultFeatureFlags = { dynamicAgentRouting: false, claudeBackgroundPromptHandoff: false, claudeMidTurnSteering: false, + officeSessionIdentity: false, } as const; export type FeatureName = keyof typeof defaultFeatureFlags; diff --git a/docs/public/configuration.md b/docs/public/configuration.md index b3cc4a5511..0e375f0905 100644 --- a/docs/public/configuration.md +++ b/docs/public/configuration.md @@ -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 diff --git a/docs/public/operations.md b/docs/public/operations.md index f0f407f43b..18b723081a 100644 --- a/docs/public/operations.md +++ b/docs/public/operations.md @@ -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.