diff --git a/apps/backend/internal/agent/runtime/lifecycle/event_types.go b/apps/backend/internal/agent/runtime/lifecycle/event_types.go index 19b1da51a1..56895073d4 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/event_types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/event_types.go @@ -264,13 +264,14 @@ type AgentStreamEventData struct { // for execution-scoped logic (e.g., resume-token CAS that must reject writes from // a defunct execution). type AgentStreamEventPayload struct { - Type string `json:"type"` // Always "agent/event" - Timestamp string `json:"timestamp"` - AgentID string `json:"agent_id"` // Historical: execution.ID. Prefer ExecutionID. - ExecutionID string `json:"execution_id"` // Lifecycle execution ID; stable across the payload's lifetime. - TaskID string `json:"task_id"` - SessionID string `json:"session_id"` // Task session ID - Data *AgentStreamEventData `json:"data"` + Type string `json:"type"` // Always "agent/event" + Timestamp string `json:"timestamp"` + AgentID string `json:"agent_id"` // Historical: execution.ID. Prefer ExecutionID. + ExecutionID string `json:"execution_id"` // Lifecycle execution ID; stable across the payload's lifetime. + AgentProfileID string `json:"agent_profile_id,omitempty"` // Stable Office identity (execution.officeProfileID()); the agent that is actually running, not the task's assignee. + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` // Task session ID + Data *AgentStreamEventData `json:"data"` } // GitEventType discriminates the type of git event diff --git a/apps/backend/internal/agent/runtime/lifecycle/events.go b/apps/backend/internal/agent/runtime/lifecycle/events.go index 2027b1f5fd..e264677d96 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/events.go @@ -274,13 +274,14 @@ func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, even // session_id is the task session ID (execution.SessionID) // acp_session_id in eventData is the internal agent protocol session payload := AgentStreamEventPayload{ - Type: "agent/event", - Timestamp: time.Now().UTC().Format(time.RFC3339Nano), - AgentID: execution.ID, - ExecutionID: execution.ID, - TaskID: execution.TaskID, - SessionID: execution.SessionID, - Data: eventData, + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + AgentID: execution.ID, + ExecutionID: execution.ID, + AgentProfileID: execution.officeProfileID(), + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: eventData, } busEvent := bus.NewEvent(events.AgentStream, "agent-manager", payload) diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go index d35140cb6a..022a5e1457 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go @@ -255,7 +255,8 @@ const ( MetadataKeyModelOverride = "model_override" // Office metadata keys - MetadataKeySkillManifestJSON = "skill_manifest_json" + MetadataKeySkillManifestJSON = "skill_manifest_json" + MetadataKeyOfficeAgentProfileID = "office_agent_profile_id" // SSH runtime metadata keys (per-session, except SSHWorkdirRoot which is per-profile). MetadataKeySSHHostAlias = "ssh_host_alias" @@ -376,6 +377,7 @@ var persistentMetadataKeys = map[string]bool{ MetadataKeyWorktreeBranch: true, MetadataKeyRemoteContributions: true, MetadataKeyContributionDestinations: true, + MetadataKeyOfficeAgentProfileID: true, } // persistentMetadataPrefixes lists key prefixes that should persist. @@ -392,6 +394,10 @@ var persistentMetadataPrefixes = []string{ // the second session would try to attach to the first session's agentctl // process and end up sharing its ACP session and instance port. var sessionScopedMetadataKeys = map[string]bool{ + // Office identity belongs to the session that produced the runtime row and + // must not be inherited by a sibling session sharing the environment. + MetadataKeyOfficeAgentProfileID: true, + MetadataKeySSHRemoteSessionDir: true, MetadataKeySSHRemoteAgentctlPort: true, MetadataKeySSHRemoteAgentctlPID: true, diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go index f09480cb97..c528eba08e 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go @@ -19,6 +19,7 @@ func TestShouldPersistMetadataKey(t *testing.T) { {name: "exact match executor_profile_id", key: "executor_profile_id", want: true}, {name: "exact match image_tag_override", key: MetadataKeyImageTagOverride, want: true}, {name: "exact match container_id", key: MetadataKeyContainerID, want: true}, + {name: "exact match office agent identity", key: MetadataKeyOfficeAgentProfileID, want: true}, {name: "prefix env_secret_id_", key: "env_secret_id_SPRITES_API_TOKEN", want: true}, {name: "prefix env_secret_id_ another key", key: "env_secret_id_OPENAI_KEY", want: true}, {name: "not persistent task_description", key: "task_description", want: false}, @@ -72,6 +73,13 @@ func TestKubernetesRuntimeMetadataKeysPersistWithoutLocalForward(t *testing.T) { "local forwards are process-local and must rotate after restart") } +func TestOfficeAgentIdentityMetadataIsSessionScoped(t *testing.T) { + require.True(t, ShouldPersistMetadataKey(MetadataKeyOfficeAgentProfileID), + "Office identity must survive same-session restart") + require.True(t, IsSessionScopedMetadataKey(MetadataKeyOfficeAgentProfileID), + "Office identity must not leak to a sibling session") +} + func TestFilterPersistentMetadata(t *testing.T) { t.Run("nil input returns nil", func(t *testing.T) { require.Nil(t, FilterPersistentMetadata(nil)) diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go index 4a76996572..cb581b6213 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go @@ -149,6 +149,45 @@ func TestHandleAgentEvent_CompleteCarriesPromptTurnID(t *testing.T) { t.Fatal("no complete stream event published") } +// TestHandleAgentEvent_CompleteCarriesActingAgentOfficeIdentity pins that the +// stream event's AgentProfileID is the acting agent's own office identity +// (execution.officeProfileID()), not the concrete AgentProfileID the CLI +// happens to run under. This is what lets office/service attribute a +// session-bridged comment to the agent that actually ran the turn instead of +// the task's assignee, without depending on task_sessions.agent_profile_id +// (which only holds the acting agent when features.officeSessionIdentity is +// on — off by default in every shipped profile). +func TestHandleAgentEvent_CompleteCarriesActingAgentOfficeIdentity(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-office-identity", "task-1", "session-1") + execution.AgentProfileID = "runner-pm" + execution.OfficeAgentProfileID = "critic" + if err := mgr.executionStore.Add(execution); err != nil { + t.Fatalf("add execution: %v", err) + } + generation, err := mgr.executionStore.BeginPrompt(execution.ID) + if err != nil { + t.Fatalf("begin prompt: %v", err) + } + + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: streams.EventTypeComplete, + SessionID: execution.SessionID, + PromptGeneration: generation, + }) + + for _, payload := range eventBus.getStreamEvents() { + if payload.Data != nil && payload.Data.Type == streams.EventTypeComplete { + if payload.AgentProfileID != "critic" { + t.Fatalf("agent_profile_id = %q, want %q (the acting agent, not %q)", + payload.AgentProfileID, "critic", execution.AgentProfileID) + } + return + } + } + t.Fatal("no complete stream event published") +} + func TestHandleAgentEvent_RecordsStreamedAssistantTextForResumeContext(t *testing.T) { mgr, _ := createTestManagerWithTracking() history, err := NewSessionHistoryManager(t.TempDir(), "", newTestLogger()) diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go index 3f09d6e4c9..e80cf591fb 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go @@ -825,6 +825,7 @@ func (m *Manager) prepareExecutionCreateRequest( } } + officeAgentProfileID := workspaceOfficeAgentProfileID(info) preparation := &executionCreatePreparation{ request: &ExecutorCreateRequest{ InstanceID: executionID, @@ -833,7 +834,7 @@ func (m *Manager) prepareExecutionCreateRequest( TaskEnvironmentID: info.TaskEnvironmentID, WorkspaceReuseRequired: info.TaskEnvironmentID != "", AgentProfileID: executionProfileID, - OfficeAgentProfileID: info.AgentProfileID, + OfficeAgentProfileID: officeAgentProfileID, WorkspacePath: info.WorkspacePath, WorkspaceSourceRoots: workspaceSourceRoots(info.WorkspaceFolders, info.WorkspaceRepositories), Protocol: string(agentConfig.Runtime().Protocol), @@ -881,11 +882,12 @@ func (m *Manager) prepareExecutionEnvironment( agentConfig agents.Agent, profileInfo *AgentProfileInfo, ) (*executionEnvironmentPreparation, error) { + officeAgentProfileID := workspaceOfficeAgentProfileID(info) managedReq := &LaunchRequest{ TaskID: taskID, WorkspaceID: info.WorkspaceID, SessionID: info.SessionID, - AgentProfileID: info.AgentProfileID, + AgentProfileID: officeAgentProfileID, ExecutionProfileID: executionProfileID, ExecutorType: info.ExecutorType, Env: make(map[string]string), @@ -1017,6 +1019,18 @@ func workspaceExecutionProfileID(info *WorkspaceInfo) string { return info.AgentProfileID } +func workspaceOfficeAgentProfileID(info *WorkspaceInfo) string { + if info == nil { + return "" + } + if value, ok := info.Metadata[MetadataKeyOfficeAgentProfileID].(string); ok { + if profileID := strings.TrimSpace(value); profileID != "" { + return profileID + } + } + return info.AgentProfileID +} + func workspaceExecutorProfileID(info *WorkspaceInfo) string { if info == nil { return "" diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_identity_recovery_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_identity_recovery_test.go new file mode 100644 index 0000000000..c5553be7d1 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_identity_recovery_test.go @@ -0,0 +1,47 @@ +package lifecycle + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestLaunchPromotesWorkspaceOnlyExecutionWithRequestedOfficeIdentity(t *testing.T) { + mgr := newTestManager(t) + mgr.profileResolver = &countingProfileResolver{info: &AgentProfileInfo{ + ProfileID: "execution-profile", + AgentName: "auggie", + }} + + existing := &AgentExecution{ + ID: "execution-workspace-only", + SessionID: "session-reviewer", + TaskID: "task-reviewer", + AgentProfileID: "execution-profile", + OfficeAgentProfileID: "assignee", + } + require.NoError(t, mgr.executionStore.Add(existing)) + + got, err := mgr.Launch(context.Background(), &LaunchRequest{ + TaskID: existing.TaskID, + SessionID: existing.SessionID, + AgentProfileID: "reviewer", + ExecutionProfileID: "execution-profile", + ACPSessionID: "acp-reviewer", + }) + require.NoError(t, err) + require.Same(t, existing, got) + require.Equal(t, "reviewer", got.OfficeAgentProfileID, + "promotion must replace the stale session owner with the acting Office identity") +} + +func TestWorkspaceOfficeAgentProfileIDPrefersPersistedIdentity(t *testing.T) { + got := workspaceOfficeAgentProfileID(&WorkspaceInfo{ + AgentProfileID: "assignee", + Metadata: map[string]interface{}{ + MetadataKeyOfficeAgentProfileID: "reviewer", + }, + }) + require.Equal(t, "reviewer", got) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go b/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go index e38dac7dfb..7b8bc8cb64 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go @@ -1437,6 +1437,16 @@ func (m *Manager) promoteWorkspaceExecution(ctx context.Context, execution *Agen if execution.AgentCommand != "" { return nil, nil } + // Workspace-only executions can be created from a session row that stores + // the task assignee. The launch request carries the acting Office identity, + // so refresh it before the execution starts emitting events. + if req.AgentProfileID != "" { + execution.OfficeAgentProfileID = req.AgentProfileID + // Persist the acting identity while the workspace-only execution is + // being promoted, so a restart before the first stream event can + // restore the same attribution. + m.persistExecutorRunning(context.WithoutCancel(sharedCtx), execution) + } agentTypeName, profileInfo, err := m.resolveAgentProfile(sharedCtx, req) if err != nil { return nil, err diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_streaming.go b/apps/backend/internal/agent/runtime/lifecycle/manager_streaming.go index 65e01b2988..490b264158 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_streaming.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_streaming.go @@ -281,13 +281,14 @@ func (m *Manager) publishStreamingContentNow( } payload := &AgentStreamEventPayload{ - Type: "agent/event", - Timestamp: time.Now().UTC().Format(time.RFC3339Nano), - AgentID: execution.ID, - ExecutionID: execution.ID, - TaskID: execution.TaskID, - SessionID: execution.SessionID, - Data: &event, + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + AgentID: execution.ID, + ExecutionID: execution.ID, + AgentProfileID: execution.officeProfileID(), + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: &event, } m.eventPublisher.PublishAgentStreamEventPayload(payload) } diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence.go b/apps/backend/internal/agent/runtime/lifecycle/persistence.go index 14c2b000a2..2121fb6e14 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence.go @@ -93,6 +93,12 @@ func buildRunningFromExecution(execution *AgentExecution, prior *models.Executor Metadata: FilterPersistentMetadata(metadata), LastSeenAt: lastSeenAt, } + if officeProfileID := strings.TrimSpace(execution.OfficeAgentProfileID); officeProfileID != "" { + if running.Metadata == nil { + running.Metadata = make(map[string]interface{}) + } + running.Metadata[MetadataKeyOfficeAgentProfileID] = officeProfileID + } if prior != nil { if strings.TrimSpace(prior.ExecutorID) != "" { running.ExecutorID = prior.ExecutorID diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go b/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go index bb66aa3c3a..a0eba1acdf 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go @@ -80,6 +80,19 @@ func TestBuildRunningFromExecutionPersistsFreshExecutorIdentity(t *testing.T) { } } +func TestBuildRunningFromExecutionPersistsOfficeAgentIdentity(t *testing.T) { + running := buildRunningFromExecution(&AgentExecution{ + ID: "exec-1", + TaskID: "task-1", + SessionID: "session-1", + OfficeAgentProfileID: "reviewer", + }, nil) + + if got := getMetadataString(running.Metadata, MetadataKeyOfficeAgentProfileID); got != "reviewer" { + t.Fatalf("persisted OfficeAgentProfileID = %q, want reviewer", got) + } +} + func TestBuildRunningFromExecutionPreservesPriorExecutorIdentity(t *testing.T) { running := buildRunningFromExecution(&AgentExecution{ ID: "exec-2", TaskID: "task-1", SessionID: "session-1", diff --git a/apps/backend/internal/office/service/comments.go b/apps/backend/internal/office/service/comments.go index 6a5c852b81..870ffbe38b 100644 --- a/apps/backend/internal/office/service/comments.go +++ b/apps/backend/internal/office/service/comments.go @@ -47,6 +47,7 @@ func (s *Service) publishCommentCreated(ctx context.Context, comment *models.Tas CommentID: comment.ID, AuthorID: comment.AuthorID, AuthorType: comment.AuthorType, + Source: comment.Source, } event := bus.NewEvent(events.OfficeCommentCreated, "office-service", data) if err := s.eb.Publish(ctx, events.OfficeCommentCreated, event); err != nil { diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 1403936365..0cc2656756 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -122,6 +122,7 @@ type CommentPostedData struct { AuthorType string `json:"author_type"` AssigneeAgentProfileID string `json:"assignee_agent_profile_id"` EngineDispatched string `json:"engine_dispatched"` + Source string `json:"source"` } // ApprovalResolvedData represents an approval resolved event payload. @@ -275,11 +276,12 @@ func (s *Service) RegisterEventSubscribers(eb bus.EventBus) error { // AgentTurnMessageData is the payload of an agent.turn.message_saved event. type AgentTurnMessageData struct { - TaskID string `json:"task_id"` - SessionID string `json:"session_id"` - TurnID string `json:"turn_id"` - AgentText string `json:"agent_text"` - AgentID string `json:"agent_id"` + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` + TurnID string `json:"turn_id"` + AgentText string `json:"agent_text"` + AgentID string `json:"agent_id"` + AgentProfileID string `json:"agent_profile_id"` } // handleAgentTurnMessageSaved auto-bridges an agent session response to a @@ -331,10 +333,32 @@ func (s *Service) handleAgentTurnMessageSaved(ctx context.Context, event *bus.Ev return nil } + // Attribute to the agent that actually ran the turn, not the task's + // assignee — the two diverge for a reviewer/approver turn. The event + // carries the acting agent's own office identity directly + // (execution.officeProfileID(), captured at launch before step/routing + // overrides mutate the profile). The session-row lookup only reflects + // the acting agent when features.officeSessionIdentity is on — off by + // default in every shipped profile, it stores the assignee for every + // participant's session — so it is kept only as a fallback for events + // published before this field existed. Final fallback is the assignee, + // mirroring handlePromptUsage's log-and-continue fallback below. + authorID := fields.AssigneeAgentProfileID + if data.AgentProfileID != "" { + authorID = data.AgentProfileID + } else if id, lookupErr := s.repo.GetSessionAgentProfileID(ctx, data.TaskID, data.SessionID); lookupErr != nil { + s.logger.Warn("session agent profile lookup failed", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.Error(lookupErr)) + } else if id != "" { + authorID = id + } + comment := &models.TaskComment{ TaskID: data.TaskID, AuthorType: "agent", - AuthorID: fields.AssigneeAgentProfileID, + AuthorID: authorID, Body: agentText, Source: "session", } @@ -344,10 +368,10 @@ func (s *Service) handleAgentTurnMessageSaved(ctx context.Context, event *bus.Ev return cErr } s.publishCommentCreated(ctx, comment) - // Successful turn → reset the agent's consecutive-failure counter - // regardless of which task succeeded. A bridged comment is the - // only place we know a turn produced real output. - s.RecordAgentSuccess(ctx, fields.AssigneeAgentProfileID) + // Successful turn → reset the acting agent's consecutive-failure + // counter. A bridged comment is the only place we know a turn + // produced real output. + s.RecordAgentSuccess(ctx, authorID) // Lifecycle: each successful turn under an in-flight run lands a // "step" event so the run detail page's Events log gets a row per // turn. Resolve the run from the currently-claimed run for this @@ -1158,6 +1182,11 @@ func (s *Service) queueCommentRun(ctx context.Context, data CommentPostedData) e if data.EngineDispatched == commentkeys.EngineDispatchedValue { return nil } + // A session-bridged comment mirrors a turn that already ran; it must + // never itself queue a new run, whichever agent it's attributed to. + if data.Source == "session" { + return nil + } // Self-comment short-circuit: if the agent that wrote the comment is // also the task's primary participant we don't want to queue a run. // The engine's queue_run resolver enforces the same rule, but diff --git a/apps/backend/internal/office/service/event_subscribers_session_attribution_test.go b/apps/backend/internal/office/service/event_subscribers_session_attribution_test.go new file mode 100644 index 0000000000..5412179bad --- /dev/null +++ b/apps/backend/internal/office/service/event_subscribers_session_attribution_test.go @@ -0,0 +1,245 @@ +package service_test + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/office/service" +) + +// These tests use DISTINCT assignee and acting-agent profiles, the blind +// spot that let the assignee-attribution bug ship: any test where the two +// are the same profile passes whether or not attribution actually runs. +// +// The primary attribution path is the event's own agent_profile_id +// (threaded from execution.officeProfileID() at the point the agent +// actually ran — see lifecycle.AgentStreamEventPayload.AgentProfileID). The +// task_sessions.agent_profile_id lookup is a fallback for events published +// before that field existed: it is NOT reliable in production, because +// officeSessionOwnerID only stores the acting agent's id there when +// features.officeSessionIdentity is on, which defaults to off in every +// shipped profile — off, that column holds the assignee for every +// participant's session row. TestAutoPostAgentComment_AttributesFromEventPayloadEvenWithStaleSessionRow +// pins exactly that: the fix must not depend on the session row being +// correct. + +// TestAutoPostAgentComment_AttributesFromEventPayloadEvenWithStaleSessionRow +// pins the core regression. The task_sessions row is seeded with the +// ASSIGNEE's id (reproducing officeSessionOwnerID's behavior with +// features.officeSessionIdentity off, the default in every shipped +// profile), but the event carries the real acting agent's profile id +// directly. Attribution must come from the event, not the stale row. +func TestAutoPostAgentComment_AttributesFromEventPayloadEvenWithStaleSessionRow(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "runner-pm") + createTestAgent(t, svc, "ws-1", "critic") + taskID := createOfficeTask(t, svc, "ws-1", "runner-pm") + // Reproduces production with officeSessionIdentity off: the session row + // holds the assignee, not the agent that is actually running. + insertTestTaskSession(t, svc, "sess-critic", taskID, "runner-pm") + + event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{ + "task_id": taskID, + "session_id": "sess-critic", + "agent_text": "Rejected: the note isn't retrievable.", + "agent_id": "exec-1234", // execution id, must NOT be used as author + "agent_profile_id": "critic", + }) + if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { + t.Fatalf("publish event: %v", err) + } + + comments, err := svc.ListComments(ctx, taskID) + if err != nil { + t.Fatalf("list comments: %v", err) + } + for _, c := range comments { + if c.Source == "session" && c.Body == "Rejected: the note isn't retrievable." { + if c.AuthorID != "critic" { + t.Fatalf("author_id = %q, want %q (the agent that actually spoke)", c.AuthorID, "critic") + } + return + } + } + t.Fatalf("expected session comment, got %+v", comments) +} + +// TestAutoPostAgentComment_AttributesActingAgentNotAssignee pins the +// legacy-fallback path: an event with no agent_profile_id (published before +// the field existed) still resolves the acting agent from +// task_sessions.agent_profile_id, not the assignee. +func TestAutoPostAgentComment_AttributesActingAgentNotAssignee(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "runner-pm") + createTestAgent(t, svc, "ws-1", "critic") + taskID := createOfficeTask(t, svc, "ws-1", "runner-pm") + insertTestTaskSession(t, svc, "sess-critic", taskID, "critic") + + event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{ + "task_id": taskID, + "session_id": "sess-critic", + "agent_text": "Rejected: the note isn't retrievable.", + "agent_id": "exec-1234", // execution id, must NOT be used as author + }) + if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { + t.Fatalf("publish event: %v", err) + } + + comments, err := svc.ListComments(ctx, taskID) + if err != nil { + t.Fatalf("list comments: %v", err) + } + for _, c := range comments { + if c.Source == "session" && c.Body == "Rejected: the note isn't retrievable." { + if c.AuthorID != "critic" { + t.Fatalf("author_id = %q, want %q (the agent that actually spoke)", c.AuthorID, "critic") + } + return + } + } + t.Fatalf("expected session comment, got %+v", comments) +} + +// TestAutoPostAgentComment_FallsBackToAssigneeWhenSessionUnresolvable pins +// the no-regression case: when neither the event nor the session row +// resolves to an agent (no agent_profile_id, missing task_sessions row), the +// bridge still falls back to the assignee instead of dropping the comment or +// leaving it unattributed. +func TestAutoPostAgentComment_FallsBackToAssigneeWhenSessionUnresolvable(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "runner-pm2") + taskID := createOfficeTask(t, svc, "ws-1", "runner-pm2") + + event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{ + "task_id": taskID, + "session_id": "sess-missing", + "agent_text": "Note delivered above.", + "agent_id": "exec-5678", + }) + if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { + t.Fatalf("publish event: %v", err) + } + + comments, err := svc.ListComments(ctx, taskID) + if err != nil { + t.Fatalf("list comments: %v", err) + } + for _, c := range comments { + if c.Source == "session" && c.Body == "Note delivered above." { + if c.AuthorID != "runner-pm2" { + t.Fatalf("author_id = %q, want %q (fallback to assignee)", c.AuthorID, "runner-pm2") + } + return + } + } + t.Fatalf("expected session comment, got %+v", comments) +} + +// TestAutoPostAgentComment_NonAssigneeAuthorQueuesNoRun pins the coupling +// documented at queueCommentRun: a session-bridged mirror of a turn must +// never itself queue a run, even once its author is the acting (non- +// assignee) agent. Before the fix this held by accident, because the +// bridge always wrote the assignee as author and tripped the self-comment +// gate; fixing the author alone (without the explicit source guard) would +// start waking the runner on every reviewer/approver turn. +func TestAutoPostAgentComment_NonAssigneeAuthorQueuesNoRun(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "runner-pm3") + createTestAgent(t, svc, "ws-1", "critic3") + taskID := createOfficeTask(t, svc, "ws-1", "runner-pm3") + insertTestTaskSession(t, svc, "sess-critic3", taskID, "critic3") + + event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{ + "task_id": taskID, + "session_id": "sess-critic3", + "agent_text": "Approved. The transition was applied.", + "agent_id": "exec-9999", + }) + if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { + t.Fatalf("publish event: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + for _, run := range runs { + if run.Reason == service.RunReasonTaskComment { + t.Fatalf("session-bridged comment must not queue a run, got %+v", run) + } + } +} + +// TestAutoPostAgentComment_RecordsSuccessForActingAgent pins that a +// successful turn's consecutive-failure reset credits the agent that +// actually spoke, not the assignee — the same resolved identity used for +// attribution (event_subscribers.go), not a second, independently-wrong +// value. The session row is seeded with the assignee's id (as +// officeSessionOwnerID leaves it with features.officeSessionIdentity off) to +// prove the reset is driven by the event's agent_profile_id, not that row. +func TestAutoPostAgentComment_RecordsSuccessForActingAgent(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "runner-pm4") + createTestAgent(t, svc, "ws-1", "critic4") + + // Give both agents prior failures so a reset is observable and + // asymmetry (only the acting agent resets) is provable. + for i := 0; i < 2; i++ { + taskID := uuidish("task-fail-runner", i) + insertSyntheticTask(t, svc, taskID, "ws-1", "runner-pm4") + w := queueAndReadRun(t, svc, "runner-pm4", taskID) + if err := svc.HandleAgentFailure(ctx, w, "boom"); err != nil { + t.Fatalf("handle failure (runner) %d: %v", i, err) + } + } + for i := 0; i < 2; i++ { + taskID := uuidish("task-fail-critic", i) + insertSyntheticTask(t, svc, taskID, "ws-1", "critic4") + w := queueAndReadRun(t, svc, "critic4", taskID) + if err := svc.HandleAgentFailure(ctx, w, "boom"); err != nil { + t.Fatalf("handle failure (critic) %d: %v", i, err) + } + } + + taskID := createOfficeTask(t, svc, "ws-1", "runner-pm4") + insertTestTaskSession(t, svc, "sess-critic4", taskID, "runner-pm4") + + event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{ + "task_id": taskID, + "session_id": "sess-critic4", + "agent_text": "Approved. All checks satisfied.", + "agent_id": "exec-4321", + "agent_profile_id": "critic4", + }) + if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { + t.Fatalf("publish event: %v", err) + } + + critic, err := svc.GetAgentInstance(ctx, "critic4") + if err != nil { + t.Fatalf("get critic: %v", err) + } + if critic.ConsecutiveFailures != 0 { + t.Fatalf("critic4 consecutive failures = %d, want 0 (acting agent's turn succeeded)", critic.ConsecutiveFailures) + } + + runner, err := svc.GetAgentInstance(ctx, "runner-pm4") + if err != nil { + t.Fatalf("get runner: %v", err) + } + if runner.ConsecutiveFailures != 2 { + t.Fatalf("runner-pm4 consecutive failures = %d, want 2 (assignee did not act, must not be credited)", runner.ConsecutiveFailures) + } +} diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 2757262a23..c1acc3db60 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -1060,6 +1060,10 @@ func (s *Service) relaunchDynamicTaskAfterFailure( } s.completeTurnForSession(ctx, data.SessionID) s.retireExecutionActivityAndPublish(ctx, data.TaskID, data.SessionID, data.AgentExecutionID) + officeAgentProfileID := data.AgentProfileID + if officeAgentProfileID == "" { + officeAgentProfileID = session.AgentProfileID + } isOfficeTask, officeErr := s.lookupOfficeTask(ctx, data.TaskID) if officeErr == nil && !isOfficeTask { _, err := s.StartCreatedSession( @@ -1070,7 +1074,7 @@ func (s *Service) relaunchDynamicTaskAfterFailure( } _, err = s.launchPreparedSessionWithDynamicFallback(ctx, task, data.SessionID, executor.LaunchOptions{ AgentProfileID: executionProfileID, - OfficeAgentProfileID: session.AgentProfileID, + OfficeAgentProfileID: officeAgentProfileID, ExecutorID: "", Prompt: prompt.text, StartAgent: true, diff --git a/apps/backend/internal/orchestrator/dynamic_launch_identity_test.go b/apps/backend/internal/orchestrator/dynamic_launch_identity_test.go new file mode 100644 index 0000000000..6ecbbd3b8e --- /dev/null +++ b/apps/backend/internal/orchestrator/dynamic_launch_identity_test.go @@ -0,0 +1,86 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" + "github.com/stretchr/testify/require" +) + +func TestRelaunchDynamicTaskAfterFailurePreservesActingOfficeIdentity(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-identity" + sessionID = "session-dynamic-identity" + executionID = "execution-dynamic-identity" + ) + + repo := setupTestRepo(t) + now := time.Now().UTC() + workspace := &models.Workspace{ + ID: "ws1", + Name: "Test", + OfficeWorkflowID: "wf1", + CreatedAt: now, + UpdatedAt: now, + } + require.NoError(t, repo.CreateWorkspace(ctx, workspace)) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ + ID: "wf1", + WorkspaceID: "ws1", + Name: "Test Workflow", + CreatedAt: now, + UpdatedAt: now, + })) + task := &models.Task{ + ID: taskID, + WorkspaceID: "ws1", + WorkflowID: "wf1", + Title: "Test Task", + Description: "desc", + State: v1.TaskStateInProgress, + CreatedAt: now, + UpdatedAt: now, + } + require.NoError(t, repo.CreateTask(ctx, task)) + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + State: models.TaskSessionStateRunning, + StartedAt: now, + UpdatedAt: now, + })) + session, err := repo.GetTaskSession(ctx, sessionID) + require.NoError(t, err) + session.AgentProfileID = "assignee" + require.NoError(t, repo.UpdateTaskSession(ctx, session)) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + var launchRequest *executor.LaunchAgentRequest + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = task.ToAPI() + agentManager := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launchRequest = req + return &executor.LaunchAgentResponse{AgentExecutionID: "successor-execution"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + require.True(t, svc.relaunchDynamicTaskAfterFailure(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + AgentProfileID: "reviewer", + }, "fallback-profile")) + require.NotNil(t, launchRequest) + require.Equal(t, "reviewer", launchRequest.OfficeAgentProfileID, + "dynamic successor must keep the acting Office identity from the failure event") + require.Equal(t, "fallback-profile", launchRequest.AgentProfileID) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index e9263db2c3..8a135dd2ba 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -332,6 +332,7 @@ func (s *Service) handleAgentErrorEvent(ctx context.Context, payload *lifecycle. SessionID: sessionID, AgentExecutionID: executionID, AgentID: payload.AgentID, + AgentProfileID: payload.AgentProfileID, PromptGeneration: payload.Data.PromptGeneration, ErrorMessage: payload.Data.Error, ProviderError: payload.Data.ProviderError, @@ -608,11 +609,12 @@ func (s *Service) publishAgentTurnCompleteForTurn(ctx context.Context, payload * // For streaming agents this will be empty — the subscriber falls back to // querying the last session message from the DB. data := map[string]string{ - "task_id": payload.TaskID, - "session_id": payload.SessionID, - "agent_text": payload.Data.Text, - "agent_id": payload.AgentID, - "turn_id": turnID, + metaKeyTaskID: payload.TaskID, + metaKeySessionID: payload.SessionID, + "agent_text": payload.Data.Text, + metaKeyAgentID: payload.AgentID, + metaKeyAgentProfileID: payload.AgentProfileID, + "turn_id": turnID, } event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", data) if err := s.eventBus.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_identity_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_identity_test.go new file mode 100644 index 0000000000..1995f3a21a --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_identity_test.go @@ -0,0 +1,40 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/events" + "github.com/stretchr/testify/require" +) + +func TestCompleteStreamForwardsActingAgentProfileIdentity(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-stream-identity", "session-stream-identity", "") + + eb := &recordingEventBus{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.eventBus = eb + svc.turnService = &repoTurnService{repo: repo} + firstTurn, err := svc.turnService.StartTurn(ctx, "session-stream-identity") + require.NoError(t, err) + svc.markExecutionCompleted("session-stream-identity", "execution-stream-identity") + svc.completeTurnForSession(ctx, "session-stream-identity") + + svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{ + TaskID: "task-stream-identity", + SessionID: "session-stream-identity", + ExecutionID: "execution-stream-identity", + AgentProfileID: "reviewer", + Data: &lifecycle.AgentStreamEventData{Type: agentEventComplete}, + }) + + require.Len(t, eb.events, 1) + require.Equal(t, events.AgentTurnMessageSaved, eb.events[0].subject) + data, ok := eb.events[0].event.Data.(map[string]string) + require.True(t, ok) + require.Equal(t, firstTurn.ID, data["turn_id"]) + require.Equal(t, "reviewer", data[metaKeyAgentProfileID]) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_transient.go b/apps/backend/internal/orchestrator/event_handlers_transient.go index 0da9873e65..45c88c7347 100644 --- a/apps/backend/internal/orchestrator/event_handlers_transient.go +++ b/apps/backend/internal/orchestrator/event_handlers_transient.go @@ -35,6 +35,7 @@ const ( metaKeySessionID = "session_id" metaKeyTaskID = "task_id" metaKeyNewState = "new_state" + metaKeyAgentID = "agent_id" metaKeyAgentProfileID = "agent_profile_id" metaKeyUpdatedAt = "updated_at" ) diff --git a/apps/backend/internal/orchestrator/executor/executor_environment_test.go b/apps/backend/internal/orchestrator/executor/executor_environment_test.go index 911ec52fc3..9f89b1d38b 100644 --- a/apps/backend/internal/orchestrator/executor/executor_environment_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_environment_test.go @@ -1004,7 +1004,7 @@ func TestReuseExistingEnvironment_FreshRepoRecoveryDropsContainerHandle(t *testi // TestApplyExecutorRunningMetadata_SkipsSessionScopedKeys pins the guard // that prevents a SECOND session on the same task from inheriting the FIRST // session's session-scoped runtime resources — agentctl PID/port, remote -// session dir, local forward port. Without this filter, the SSH executor's +// session dir, local forward port, and Office agent identity. Without this filter, the SSH executor's // ResumeRemoteInstance would interpret those keys as a resume hint and // reattach to session-1's agentctl process, so session 2 would end up // sharing session 1's ACP session and instance port and never finish its @@ -1032,6 +1032,7 @@ func TestApplyExecutorRunningMetadata_SkipsSessionScopedKeys(t *testing.T) { lifecycle.MetadataKeySSHRemoteAgentctlPID: "12345", lifecycle.MetadataKeySSHLocalForwardPort: "59123", lifecycle.MetadataKeySSHRemoteAgentctlURL: "http://127.0.0.1:59123", + lifecycle.MetadataKeyOfficeAgentProfileID: "reviewer", // Non-persistent key — must NOT propagate (not in persistentMetadataKeys). "task_description": "session 1 prompt", }, @@ -1067,6 +1068,7 @@ func TestApplyExecutorRunningMetadata_SkipsSessionScopedKeys(t *testing.T) { lifecycle.MetadataKeySSHRemoteAgentctlPID, lifecycle.MetadataKeySSHLocalForwardPort, lifecycle.MetadataKeySSHRemoteAgentctlURL, + lifecycle.MetadataKeyOfficeAgentProfileID, } for _, k := range sessionScoped { if v, ok := req.Metadata[k]; ok { diff --git a/apps/backend/internal/task/service/service_turns.go b/apps/backend/internal/task/service/service_turns.go index 4b16585f34..2a26def904 100644 --- a/apps/backend/internal/task/service/service_turns.go +++ b/apps/backend/internal/task/service/service_turns.go @@ -906,6 +906,9 @@ func (s *Service) GetWorkspaceInfoForSession(ctx context.Context, taskID, sessio info.RuntimeName = running.Runtime info.AgentExecutionID = running.AgentExecutionID mergePersistentWorkspaceMetadata(info, running.Metadata) + if officeProfileID, ok := running.Metadata[lifecycle.MetadataKeyOfficeAgentProfileID].(string); ok && strings.TrimSpace(officeProfileID) != "" { + info.AgentProfileID = officeProfileID + } if running.ContainerID != "" { ensureWorkspaceMetadata(info)[lifecycle.MetadataKeyContainerID] = running.ContainerID } diff --git a/apps/backend/internal/task/service/service_turns_test.go b/apps/backend/internal/task/service/service_turns_test.go index 850ac962fb..d1bef693a3 100644 --- a/apps/backend/internal/task/service/service_turns_test.go +++ b/apps/backend/internal/task/service/service_turns_test.go @@ -895,6 +895,47 @@ func TestGetWorkspaceInfoForSession_BasicFields(t *testing.T) { } } +func TestGetWorkspaceInfoForSessionRestoresOfficeAgentIdentityFromRuntimeInventory(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + setupTestTask(t, repo) + now := time.Now().UTC() + sessionID := "session-office-recovery" + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: "task-123", + AgentProfileID: "assignee", + ExecutionProfileID: "execution-profile", + State: models.TaskSessionStateRunning, + StartedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: sessionID, + SessionID: sessionID, + TaskID: "task-123", + ExecutionProfileID: "execution-profile", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusReady, + AgentExecutionID: "execution-office-recovery", + Metadata: map[string]interface{}{ + lifecycle.MetadataKeyOfficeAgentProfileID: "reviewer", + }, + }); err != nil { + t.Fatalf("UpsertExecutorRunning: %v", err) + } + + info, err := svc.GetWorkspaceInfoForSession(ctx, "task-123", sessionID) + if err != nil { + t.Fatalf("GetWorkspaceInfoForSession: %v", err) + } + if info.AgentProfileID != "reviewer" { + t.Fatalf("AgentProfileID = %q, want reviewer from runtime inventory", info.AgentProfileID) + } +} + func TestApplyTaskEnvironmentToWorkspaceInfoUsesEnvironmentProfileAsFallback(t *testing.T) { info := &lifecycle.WorkspaceInfo{} applyTaskEnvironmentToWorkspaceInfo(info, &models.TaskEnvironment{ diff --git a/apps/web/e2e/tests/review/submodule-review.spec.ts b/apps/web/e2e/tests/review/submodule-review.spec.ts index e0cff14081..e489e568d2 100644 --- a/apps/web/e2e/tests/review/submodule-review.spec.ts +++ b/apps/web/e2e/tests/review/submodule-review.spec.ts @@ -87,7 +87,18 @@ test.describe("Nested submodule Review", () => { await expect .poll(() => session.reviewDiffText(), { timeout: 45_000 }) .toEqual(expect.stringContaining("parent working-tree change")); - for (const expected of ["outer committed change", "inner committed change"]) { + for (const [repositoryName, expected] of [ + ["vendor/outer", "outer committed change"], + ["vendor/outer/vendor/inner", "inner committed change"], + ] as const) { + // Diff bodies load when they enter the viewport. Select each nested + // file in the tree so the review list scrolls to it before asserting + // its content, just as a reviewer would navigate the changes. + await review + .locator( + `[data-testid="review-file-row"][data-file-path="README.md"][data-repository-name="${repositoryName}"]`, + ) + .click(); await expect.poll(() => session.reviewDiffText(), { timeout: 45_000 }).toContain(expected); } await expectStickyReviewHeaderClearance(review, "mouse");