refactor(orchestrator): extract WatcherDispatchCoordinator + WatcherSource (#1070)

apps/backend/AGENTS.md

@@ -116,6 +116,8 @@ apps/backend/
 - Handles event-driven state transitions via workflow engine
 - Located in `internal/orchestrator/`
 
+**Watcher Dispatch Coordinator** (`internal/orchestrator/watcher_dispatch.go`) is the single pipeline that turns a freshly-observed external issue (Linear, Jira, future) into a Kandev task. Bus subscribers for each integration forward the event to `WatcherDispatchCoordinator.Dispatch` with a per-integration `WatcherSource` implementation (`source_linear.go`, `source_jira.go`). Source methods carry the integration-specific bits (reserve dedup, build task request, attach task ID, release, auto-start params); the coordinator owns the cross-cutting pipeline (create task, decide auto-start, error/release handling). Add a new watcher = implement `WatcherSource` + register a one-line bus subscriber. Do NOT add another `createXIssueTask` mirror.
+
 **Workflow Engine** (`internal/workflow/engine/`) provides typed state-machine evaluation:
 - `Engine.HandleTrigger()` evaluates step actions for triggers (on_enter, on_turn_start, on_turn_complete, on_exit)
 - `TransitionStore` interface abstracts persistence (implemented by `orchestrator.workflowStore`)

apps/backend/internal/orchestrator/event_handlers_github.go

@@ -126,6 +126,7 @@ func (s *Service) SetRepositoryResolver(rr RepositoryResolver) {
 // SetIssueTaskCreator sets the task creator for issue watch auto-task creation.
 func (s *Service) SetIssueTaskCreator(tc IssueTaskCreator) {
 	s.issueTaskCreator = tc
+	s.initWatcherCoordinator()
 }
 
 // handlePRFeedback logs PR feedback events. WS broadcasting is handled in main.go.

apps/backend/internal/orchestrator/event_handlers_jira.go

@@ -2,7 +2,6 @@ package orchestrator
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
 	"go.uber.org/zap"
@@ -40,128 +39,20 @@ func (s *Service) subscribeJiraEvents() {
 	}
 }
 
-// handleNewJiraIssue creates a Kandev task for a freshly-observed JIRA ticket.
-// Reuses the same IssueTaskCreator the GitHub flow uses so task creation logic
-// (workflow placement, on-enter actions, etc.) stays in one place.
-func (s *Service) handleNewJiraIssue(_ context.Context, event *bus.Event) error {
+// handleNewJiraIssue translates a JiraNewIssue bus event into a
+// dispatchWatcherEvent call. Symmetric with handleNewLinearIssue.
+func (s *Service) handleNewJiraIssue(ctx context.Context, event *bus.Event) error {
 	evt, ok := event.Data.(*jira.NewJiraIssueEvent)
-	if !ok {
+	if !ok || evt == nil || evt.Issue == nil {
 		return nil
 	}
-	if evt.Issue == nil {
-		return nil
-	}
-	s.logger.Info("new jira issue detected from watch",
+	s.dispatchWatcherEvent(ctx, "jira",
+		NewJiraWatcherSource(s.jiraService, s.logger), evt,
 		zap.String("issue_watch_id", evt.IssueWatchID),
 		zap.String("issue_key", evt.Issue.Key))
-
-	if s.issueTaskCreator == nil {
-		s.logger.Warn("issue task creator not configured, skipping jira task creation")
-		return nil
-	}
-
-	// Background context: the bus delivery context may be cancelled before
-	// task creation finishes. The work is independent of the publisher.
-	go s.createJiraIssueTask(context.Background(), evt)
 	return nil
 }
 
-func (s *Service) createJiraIssueTask(ctx context.Context, evt *jira.NewJiraIssueEvent) {
-	ticket := evt.Issue
-	if !s.reserveJiraIssue(ctx, evt) {
-		return
-	}
-
-	req := &IssueTaskRequest{
-		WorkspaceID:    evt.WorkspaceID,
-		WorkflowID:     evt.WorkflowID,
-		WorkflowStepID: evt.WorkflowStepID,
-		Title:          fmt.Sprintf("[%s] %s", ticket.Key, ticket.Summary),
-		Description:    interpolateJiraPrompt(evt.Prompt, ticket),
-		Metadata: map[string]interface{}{
-			"jira_issue_watch_id": evt.IssueWatchID,
-			"jira_issue_key":      ticket.Key,
-			"jira_issue_url":      ticket.URL,
-			"jira_status":         ticket.StatusName,
-			"jira_assignee":       ticket.AssigneeName,
-			"agent_profile_id":    evt.AgentProfileID,
-			"executor_profile_id": evt.ExecutorProfileID,
-		},
-	}
-
-	task, err := s.issueTaskCreator.CreateIssueTask(ctx, req)
-	if err != nil {
-		s.logger.Error("failed to create jira issue task",
-			zap.String("issue_watch_id", evt.IssueWatchID),
-			zap.String("issue_key", ticket.Key),
-			zap.Error(err))
-		s.releaseJiraIssue(ctx, evt)
-		return
-	}
-
-	s.attachJiraIssueTaskID(ctx, evt, task.ID)
-	s.logger.Info("created jira issue task",
-		zap.String("task_id", task.ID),
-		zap.String("issue_key", ticket.Key))
-
-	if !s.shouldAutoStartStep(ctx, evt.WorkflowStepID) {
-		return
-	}
-	if _, err := s.StartTask(
-		ctx, task.ID, evt.AgentProfileID, "", evt.ExecutorProfileID,
-		"", task.Description, evt.WorkflowStepID, false, true, nil,
-	); err != nil {
-		s.logger.Error("failed to auto-start jira issue task",
-			zap.String("task_id", task.ID), zap.Error(err))
-		return
-	}
-	s.logger.Info("auto-started jira issue task",
-		zap.String("task_id", task.ID),
-		zap.String("issue_key", ticket.Key))
-}
-
-// reserveJiraIssue claims the dedup slot before the (relatively expensive)
-// task creation. Returns false if another handler beat us to it. When the
-// jira service isn't wired (boot order corner case), proceed anyway — better
-// to risk a duplicate task than silently drop the event.
-func (s *Service) reserveJiraIssue(ctx context.Context, evt *jira.NewJiraIssueEvent) bool {
-	if s.jiraService == nil {
-		return true
-	}
-	reserved, err := s.jiraService.ReserveIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Key, evt.Issue.URL)
-	if err != nil {
-		s.logger.Error("failed to reserve jira issue slot",
-			zap.String("issue_key", evt.Issue.Key), zap.Error(err))
-		return false
-	}
-	if !reserved {
-		s.logger.Debug("jira issue already reserved by concurrent handler, skipping",
-			zap.String("issue_key", evt.Issue.Key))
-	}
-	return reserved
-}
-
-func (s *Service) releaseJiraIssue(ctx context.Context, evt *jira.NewJiraIssueEvent) {
-	if s.jiraService == nil {
-		return
-	}
-	if err := s.jiraService.ReleaseIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Key); err != nil {
-		s.logger.Warn("failed to release jira issue reservation after task-create failure",
-			zap.String("issue_key", evt.Issue.Key), zap.Error(err))
-	}
-}
-
-func (s *Service) attachJiraIssueTaskID(ctx context.Context, evt *jira.NewJiraIssueEvent, taskID string) {
-	if s.jiraService == nil {
-		return
-	}
-	if err := s.jiraService.AssignIssueWatchTaskID(ctx, evt.IssueWatchID, evt.Issue.Key, taskID); err != nil {
-		s.logger.Error("failed to assign task ID to jira issue reservation",
-			zap.String("task_id", taskID),
-			zap.String("issue_key", evt.Issue.Key), zap.Error(err))
-	}
-}
-
 // interpolateJiraPrompt replaces {{issue.*}} placeholders with ticket fields.
 // When the template is empty, falls back to the embedded default in
 // config/prompts/jira-issue-watch-default.md — same pattern the GitHub

apps/backend/internal/orchestrator/event_handlers_jira_test.go

@@ -66,6 +66,15 @@ func setupJiraTaskTest(t *testing.T) *Service {
 	return createTestService(repo, stepGetter, newMockTaskRepo())
 }
 
+// dispatchJiraEvent drives an event through the watcher coordinator
+// synchronously using a JiraWatcherSource — mirrors what handleNewJiraIssue
+// does, but without the goroutine, so tests can assert on observable side
+// effects deterministically.
+func dispatchJiraEvent(svc *Service, evt *jira.NewJiraIssueEvent) {
+	src := NewJiraWatcherSource(svc.jiraService, svc.logger)
+	svc.watcherCoordinator.Dispatch(context.Background(), src, evt)
+}
+
 func TestCreateJiraIssueTask_HappyPath(t *testing.T) {
 	svc := setupJiraTaskTest(t)
 	jiraSvc := &mockJiraService{reserveReturn: true}
@@ -73,7 +82,7 @@ func TestCreateJiraIssueTask_HappyPath(t *testing.T) {
 	creator := &countingIssueTaskCreator{taskID: "task-jira-1"}
 	svc.SetIssueTaskCreator(creator)
 
-	svc.createJiraIssueTask(context.Background(), newJiraIssueEvent())
+	dispatchJiraEvent(svc, newJiraIssueEvent())
 
 	if jiraSvc.reserveCalls != 1 {
 		t.Errorf("expected 1 Reserve call, got %d", jiraSvc.reserveCalls)
@@ -102,7 +111,7 @@ func TestCreateJiraIssueTask_SkipsWhenAlreadyReserved(t *testing.T) {
 	creator := &countingIssueTaskCreator{}
 	svc.SetIssueTaskCreator(creator)
 
-	svc.createJiraIssueTask(context.Background(), newJiraIssueEvent())
+	dispatchJiraEvent(svc, newJiraIssueEvent())
 
 	if creator.calls != 0 {
 		t.Errorf("expected CreateIssueTask NOT to be called when reservation is lost, got %d", creator.calls)
@@ -119,7 +128,7 @@ func TestCreateJiraIssueTask_ReleasesWhenCreateFails(t *testing.T) {
 	creator := &countingIssueTaskCreator{err: errors.New("task creation failed")}
 	svc.SetIssueTaskCreator(creator)
 
-	svc.createJiraIssueTask(context.Background(), newJiraIssueEvent())
+	dispatchJiraEvent(svc, newJiraIssueEvent())
 
 	if jiraSvc.assignCalls != 0 {
 		t.Errorf("expected no Assign when task creation failed, got %d", jiraSvc.assignCalls)
@@ -129,6 +138,31 @@ func TestCreateJiraIssueTask_ReleasesWhenCreateFails(t *testing.T) {
 	}
 }
 
+// TestSetIssueTaskCreator_RefreshesCoordinatorTaskCreator pins the wiring
+// contract that calling SetIssueTaskCreator again must update the coordinator.
+// Regression guard: an earlier version of initWatcherCoordinator returned
+// early on the second call and silently kept the original creator.
+func TestSetIssueTaskCreator_RefreshesCoordinatorTaskCreator(t *testing.T) {
+	svc := setupJiraTaskTest(t)
+	jiraSvc := &mockJiraService{reserveReturn: true}
+	svc.SetJiraService(jiraSvc)
+
+	first := &countingIssueTaskCreator{taskID: "task-first"}
+	svc.SetIssueTaskCreator(first)
+
+	second := &countingIssueTaskCreator{taskID: "task-second"}
+	svc.SetIssueTaskCreator(second)
+
+	dispatchJiraEvent(svc, newJiraIssueEvent())
+
+	if first.calls != 0 {
+		t.Errorf("first creator must not be invoked after being replaced, got %d calls", first.calls)
+	}
+	if second.calls != 1 {
+		t.Errorf("expected the latest creator to be used, got %d calls", second.calls)
+	}
+}
+
 func TestInterpolateJiraPrompt(t *testing.T) {
 	ticket := &jira.JiraTicket{
 		Key:          "PROJ-7",

apps/backend/internal/orchestrator/event_handlers_linear.go

@@ -2,7 +2,6 @@ package orchestrator
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
 	"go.uber.org/zap"
@@ -11,7 +10,6 @@ import (
 	"github.com/kandev/kandev/internal/events"
 	"github.com/kandev/kandev/internal/events/bus"
 	"github.com/kandev/kandev/internal/linear"
-	"github.com/kandev/kandev/internal/task/models"
 )
 
 // LinearService is the subset of linear.Service the orchestrator needs to
@@ -38,132 +36,25 @@ func (s *Service) subscribeLinearEvents() {
 	}
 }
 
-// handleNewLinearIssue creates a Kandev task for a freshly-observed Linear issue.
-func (s *Service) handleNewLinearIssue(_ context.Context, event *bus.Event) error {
+// handleNewLinearIssue translates a LinearNewIssue bus event into a
+// dispatchWatcherEvent call. Integration-specific work
+// (reserve, build, attach, release, auto-start params) lives in
+// LinearWatcherSource; the pipeline (create, auto-start, error/release
+// handling) lives in the coordinator; the shared wiring guards
+// (creator/coordinator nil-checks, cancellation detachment) live in
+// the dispatchWatcherEvent helper.
+func (s *Service) handleNewLinearIssue(ctx context.Context, event *bus.Event) error {
 	evt, ok := event.Data.(*linear.NewLinearIssueEvent)
-	if !ok {
+	if !ok || evt == nil || evt.Issue == nil {
 		return nil
 	}
-	if evt.Issue == nil {
-		return nil
-	}
-	s.logger.Info("new linear issue detected from watch",
+	s.dispatchWatcherEvent(ctx, "linear",
+		NewLinearWatcherSource(s.linearService, s.logger), evt,
 		zap.String("issue_watch_id", evt.IssueWatchID),
 		zap.String("identifier", evt.Issue.Identifier))
-
-	if s.issueTaskCreator == nil {
-		s.logger.Warn("issue task creator not configured, skipping linear task creation")
-		return nil
-	}
-
-	// Background context: the bus delivery context may be cancelled before
-	// task creation finishes. The work is independent of the publisher.
-	go s.createLinearIssueTask(context.Background(), evt)
 	return nil
 }
 
-// createLinearIssueTask mirrors createJiraIssueTask intentionally: the dedup-
-// reserve → create → assign-or-release lifecycle is the same shape across
-// both integrations. Refactoring out the symmetry would couple two
-// integration packages we'd rather keep independent (their event payloads,
-// metadata keys, and prompt placeholders all differ).
-//
-//nolint:dupl // intentional symmetry with createJiraIssueTask; see comment above
-func (s *Service) createLinearIssueTask(ctx context.Context, evt *linear.NewLinearIssueEvent) {
-	issue := evt.Issue
-	if !s.reserveLinearIssue(ctx, evt) {
-		return
-	}
-
-	req := &IssueTaskRequest{
-		WorkspaceID:    evt.WorkspaceID,
-		WorkflowID:     evt.WorkflowID,
-		WorkflowStepID: evt.WorkflowStepID,
-		Title:          fmt.Sprintf("[%s] %s", issue.Identifier, issue.Title),
-		Description:    interpolateLinearPrompt(evt.Prompt, issue),
-		Metadata: map[string]interface{}{
-			"linear_issue_watch_id":         evt.IssueWatchID,
-			"linear_issue_identifier":       issue.Identifier,
-			"linear_issue_url":              issue.URL,
-			"linear_state":                  issue.StateName,
-			"linear_assignee":               issue.AssigneeName,
-			models.MetaKeyAgentProfileID:    evt.AgentProfileID,
-			models.MetaKeyExecutorProfileID: evt.ExecutorProfileID,
-		},
-	}
-
-	task, err := s.issueTaskCreator.CreateIssueTask(ctx, req)
-	if err != nil {
-		s.logger.Error("failed to create linear issue task",
-			zap.String("issue_watch_id", evt.IssueWatchID),
-			zap.String("identifier", issue.Identifier),
-			zap.Error(err))
-		s.releaseLinearIssue(ctx, evt)
-		return
-	}
-
-	s.attachLinearIssueTaskID(ctx, evt, task.ID)
-	s.logger.Info("created linear issue task",
-		zap.String("task_id", task.ID),
-		zap.String("identifier", issue.Identifier))
-
-	if !s.shouldAutoStartStep(ctx, evt.WorkflowStepID) {
-		return
-	}
-	if _, err := s.StartTask(
-		ctx, task.ID, evt.AgentProfileID, "", evt.ExecutorProfileID,
-		"", task.Description, evt.WorkflowStepID, false, true, nil,
-	); err != nil {
-		s.logger.Error("failed to auto-start linear issue task",
-			zap.String("task_id", task.ID), zap.Error(err))
-		return
-	}
-	s.logger.Info("auto-started linear issue task",
-		zap.String("task_id", task.ID),
-		zap.String("identifier", issue.Identifier))
-}
-
-// reserveLinearIssue claims the dedup slot before task creation. When the
-// linear service isn't wired (boot order corner case), proceed anyway —
-// better to risk a duplicate task than silently drop the event.
-func (s *Service) reserveLinearIssue(ctx context.Context, evt *linear.NewLinearIssueEvent) bool {
-	if s.linearService == nil {
-		return true
-	}
-	reserved, err := s.linearService.ReserveIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Identifier, evt.Issue.URL)
-	if err != nil {
-		s.logger.Error("failed to reserve linear issue slot",
-			zap.String("identifier", evt.Issue.Identifier), zap.Error(err))
-		return false
-	}
-	if !reserved {
-		s.logger.Debug("linear issue already reserved by concurrent handler, skipping",
-			zap.String("identifier", evt.Issue.Identifier))
-	}
-	return reserved
-}
-
-func (s *Service) releaseLinearIssue(ctx context.Context, evt *linear.NewLinearIssueEvent) {
-	if s.linearService == nil {
-		return
-	}
-	if err := s.linearService.ReleaseIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Identifier); err != nil {
-		s.logger.Warn("failed to release linear issue reservation after task-create failure",
-			zap.String("identifier", evt.Issue.Identifier), zap.Error(err))
-	}
-}
-
-func (s *Service) attachLinearIssueTaskID(ctx context.Context, evt *linear.NewLinearIssueEvent, taskID string) {
-	if s.linearService == nil {
-		return
-	}
-	if err := s.linearService.AssignIssueWatchTaskID(ctx, evt.IssueWatchID, evt.Issue.Identifier, taskID); err != nil {
-		s.logger.Error("failed to assign task ID to linear issue reservation",
-			zap.String("task_id", taskID),
-			zap.String("identifier", evt.Issue.Identifier), zap.Error(err))
-	}
-}
-
 // interpolateLinearPrompt replaces {{issue.*}} placeholders with issue fields.
 // When the template is empty (user didn't configure a custom prompt), it falls
 // back to the embedded default at config/prompts/linear-issue-watch-default.md

apps/backend/internal/orchestrator/service.go

@@ -259,6 +259,10 @@ type Service struct {
 	// Issue task creator for auto-creating tasks from issue watch events
 	issueTaskCreator IssueTaskCreator
 
+	// Watcher dispatch coordinator: single seam for the watcher→task pipeline.
+	// Constructed lazily once issueTaskCreator is wired (see SetIssueTaskCreator).
+	watcherCoordinator *WatcherDispatchCoordinator
+
 	// Jira service for issue watch dedup operations
 	jiraService JiraService
 

apps/backend/internal/orchestrator/source_jira.go

@@ -0,0 +1,96 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.uber.org/zap"
+
+	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/jira"
+)
+
+// JiraWatcherSource adapts the Jira integration onto the WatcherSource
+// pipeline. Symmetric with LinearWatcherSource; the only differences are
+// the metadata keys and the interpolation function.
+type JiraWatcherSource struct {
+	service JiraService
+	logger  *logger.Logger
+}
+
+// NewJiraWatcherSource constructs a source bound to the orchestrator's
+// Jira service handle.
+func NewJiraWatcherSource(svc JiraService, log *logger.Logger) *JiraWatcherSource {
+	return &JiraWatcherSource{service: svc, logger: log}
+}
+
+func (s *JiraWatcherSource) Name() string { return "jira" }
+
+func (s *JiraWatcherSource) Reserve(ctx context.Context, evt any) (bool, error) {
+	e, ok := evt.(*jira.NewJiraIssueEvent)
+	if !ok || e == nil || e.Issue == nil {
+		return false, errors.New("jira source: event payload missing or wrong type")
+	}
+	if s.service == nil {
+		return true, nil
+	}
+	return s.service.ReserveIssueWatchTask(ctx, e.IssueWatchID, e.Issue.Key, e.Issue.URL)
+}
+
+func (s *JiraWatcherSource) Release(ctx context.Context, evt any) {
+	e, ok := evt.(*jira.NewJiraIssueEvent)
+	if !ok || e == nil || e.Issue == nil || s.service == nil {
+		return
+	}
+	if err := s.service.ReleaseIssueWatchTask(ctx, e.IssueWatchID, e.Issue.Key); err != nil && s.logger != nil {
+		s.logger.Warn("jira source: release failed",
+			zap.String("issue_key", e.Issue.Key), zap.Error(err))
+	}
+}
+
+func (s *JiraWatcherSource) BuildTaskRequest(evt any) (*IssueTaskRequest, error) {
+	e, ok := evt.(*jira.NewJiraIssueEvent)
+	if !ok || e == nil || e.Issue == nil {
+		return nil, errors.New("jira source: event payload missing or wrong type")
+	}
+	return &IssueTaskRequest{
+		WorkspaceID:    e.WorkspaceID,
+		WorkflowID:     e.WorkflowID,
+		WorkflowStepID: e.WorkflowStepID,
+		Title:          fmt.Sprintf("[%s] %s", e.Issue.Key, e.Issue.Summary),
+		Description:    interpolateJiraPrompt(e.Prompt, e.Issue),
+		// Preserve today's literal metadata keys (do NOT normalise to
+		// models.MetaKey* in this refactor — separate cleanup).
+		Metadata: map[string]interface{}{
+			"jira_issue_watch_id": e.IssueWatchID,
+			"jira_issue_key":      e.Issue.Key,
+			"jira_issue_url":      e.Issue.URL,
+			"jira_status":         e.Issue.StatusName,
+			"jira_assignee":       e.Issue.AssigneeName,
+			"agent_profile_id":    e.AgentProfileID,
+			"executor_profile_id": e.ExecutorProfileID,
+		},
+	}, nil
+}
+
+func (s *JiraWatcherSource) AttachTaskID(ctx context.Context, evt any, taskID string) error {
+	e, ok := evt.(*jira.NewJiraIssueEvent)
+	if !ok || e == nil || e.Issue == nil || s.service == nil {
+		return nil
+	}
+	return s.service.AssignIssueWatchTaskID(ctx, e.IssueWatchID, e.Issue.Key, taskID)
+}
+
+func (s *JiraWatcherSource) AutoStartParams(evt any) AutoStartParams {
+	e, ok := evt.(*jira.NewJiraIssueEvent)
+	if !ok || e == nil {
+		return AutoStartParams{}
+	}
+	return AutoStartParams{
+		AgentProfileID:    e.AgentProfileID,
+		ExecutorProfileID: e.ExecutorProfileID,
+		Prompt:            interpolateJiraPrompt(e.Prompt, e.Issue),
+		WorkflowStepID:    e.WorkflowStepID,
+	}
+}

apps/backend/internal/orchestrator/source_jira_test.go

@@ -0,0 +1,164 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kandev/kandev/internal/jira"
+)
+
+type fakeJiraService struct {
+	reserveOK  bool
+	reserveErr error
+	assignErr  error
+	releaseErr error
+	gotReserve []string
+	gotAssign  []string
+	gotRelease []string
+}
+
+func (f *fakeJiraService) ReserveIssueWatchTask(_ context.Context, watchID, key, _ string) (bool, error) {
+	f.gotReserve = append(f.gotReserve, watchID+":"+key)
+	return f.reserveOK, f.reserveErr
+}
+
+func (f *fakeJiraService) AssignIssueWatchTaskID(_ context.Context, watchID, key, taskID string) error {
+	f.gotAssign = append(f.gotAssign, watchID+":"+key+":"+taskID)
+	return f.assignErr
+}
+
+func (f *fakeJiraService) ReleaseIssueWatchTask(_ context.Context, watchID, key string) error {
+	f.gotRelease = append(f.gotRelease, watchID+":"+key)
+	return f.releaseErr
+}
+
+func sampleJiraEvent() *jira.NewJiraIssueEvent {
+	return &jira.NewJiraIssueEvent{
+		IssueWatchID:      "watch-1",
+		WorkspaceID:       "ws-1",
+		WorkflowID:        "wf-1",
+		WorkflowStepID:    "step-1",
+		AgentProfileID:    "agent-1",
+		ExecutorProfileID: "exec-1",
+		Prompt:            "Pick up {{issue.key}}: {{issue.summary}}",
+		Issue: &jira.JiraTicket{
+			Key:          "PROJ-1",
+			Summary:      "Bug",
+			URL:          "https://example.atlassian.net/browse/PROJ-1",
+			StatusName:   "Open",
+			AssigneeName: "Alice",
+			Priority:     "High",
+			IssueType:    "Bug",
+			ReporterName: "Bob",
+			ProjectKey:   "PROJ",
+			Description:  "details",
+		},
+	}
+}
+
+func TestJiraSource_Name(t *testing.T) {
+	src := &JiraWatcherSource{}
+	if src.Name() != "jira" {
+		t.Fatalf("expected name=jira, got %q", src.Name())
+	}
+}
+
+func TestJiraSource_Reserve_Passthrough(t *testing.T) {
+	svc := &fakeJiraService{reserveOK: true}
+	src := &JiraWatcherSource{service: svc}
+	ok, err := src.Reserve(context.Background(), sampleJiraEvent())
+	if err != nil || !ok {
+		t.Fatalf("expected reserve ok, got ok=%v err=%v", ok, err)
+	}
+	if len(svc.gotReserve) != 1 || svc.gotReserve[0] != "watch-1:PROJ-1" {
+		t.Fatalf("unexpected reserve args: %v", svc.gotReserve)
+	}
+}
+
+func TestJiraSource_Reserve_NilServiceFailOpen(t *testing.T) {
+	src := &JiraWatcherSource{service: nil}
+	ok, err := src.Reserve(context.Background(), sampleJiraEvent())
+	if err != nil || !ok {
+		t.Fatalf("expected nil service to fail open, got ok=%v err=%v", ok, err)
+	}
+}
+
+func TestJiraSource_Reserve_Error(t *testing.T) {
+	svc := &fakeJiraService{reserveErr: errors.New("boom")}
+	src := &JiraWatcherSource{service: svc}
+	if _, err := src.Reserve(context.Background(), sampleJiraEvent()); err == nil {
+		t.Fatal("expected reserve error to surface")
+	}
+}
+
+func TestJiraSource_BuildTaskRequest(t *testing.T) {
+	src := &JiraWatcherSource{}
+	req, err := src.BuildTaskRequest(sampleJiraEvent())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if req.Title != "[PROJ-1] Bug" {
+		t.Errorf("title = %q", req.Title)
+	}
+	if req.Description != "Pick up PROJ-1: Bug" {
+		t.Errorf("prompt interpolation wrong: %q", req.Description)
+	}
+	if req.Metadata["jira_issue_key"] != "PROJ-1" {
+		t.Errorf("missing jira_issue_key metadata")
+	}
+	if req.Metadata["agent_profile_id"] != "agent-1" {
+		t.Errorf("missing agent_profile_id metadata")
+	}
+}
+
+func TestJiraSource_BuildTaskRequest_WrongType(t *testing.T) {
+	src := &JiraWatcherSource{}
+	if _, err := src.BuildTaskRequest("nope"); err == nil {
+		t.Fatal("expected error for wrong event type")
+	}
+}
+
+func TestJiraSource_AttachTaskID(t *testing.T) {
+	svc := &fakeJiraService{}
+	src := &JiraWatcherSource{service: svc}
+	if err := src.AttachTaskID(context.Background(), sampleJiraEvent(), "task-9"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(svc.gotAssign) != 1 || svc.gotAssign[0] != "watch-1:PROJ-1:task-9" {
+		t.Fatalf("unexpected assign args: %v", svc.gotAssign)
+	}
+}
+
+func TestJiraSource_Release(t *testing.T) {
+	svc := &fakeJiraService{}
+	src := &JiraWatcherSource{service: svc}
+	src.Release(context.Background(), sampleJiraEvent())
+	if len(svc.gotRelease) != 1 || svc.gotRelease[0] != "watch-1:PROJ-1" {
+		t.Fatalf("unexpected release args: %v", svc.gotRelease)
+	}
+}
+
+func TestJiraSource_Release_ErrorIsLoggedNotPropagated(t *testing.T) {
+	svc := &fakeJiraService{releaseErr: errors.New("dedup store down")}
+	src := &JiraWatcherSource{service: svc, logger: nopLogger(t)}
+	// Release returns no value: failure must be swallowed and logged.
+	src.Release(context.Background(), sampleJiraEvent())
+	if len(svc.gotRelease) != 1 {
+		t.Fatalf("expected release call to be attempted, got %d", len(svc.gotRelease))
+	}
+}
+
+func TestJiraSource_AutoStartParams(t *testing.T) {
+	src := &JiraWatcherSource{}
+	p := src.AutoStartParams(sampleJiraEvent())
+	if p.AgentProfileID != "agent-1" || p.ExecutorProfileID != "exec-1" {
+		t.Fatalf("unexpected auto-start params: %+v", p)
+	}
+	if p.WorkflowStepID != "step-1" {
+		t.Errorf("step id wrong: %q", p.WorkflowStepID)
+	}
+	if p.Prompt != "Pick up PROJ-1: Bug" {
+		t.Errorf("prompt should be interpolated, got %q", p.Prompt)
+	}
+}

apps/backend/internal/orchestrator/source_linear.go

@@ -0,0 +1,97 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.uber.org/zap"
+
+	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/linear"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// LinearWatcherSource adapts the Linear integration onto the WatcherSource
+// pipeline. Wraps the existing LinearService (already used by the orchestrator
+// for dedup) and the package-local interpolateLinearPrompt so behaviour is
+// preserved byte-for-byte versus the old createLinearIssueTask helper.
+type LinearWatcherSource struct {
+	service LinearService
+	logger  *logger.Logger
+}
+
+// NewLinearWatcherSource constructs a source bound to the orchestrator's
+// Linear service handle. logger may be nil — methods that log will no-op.
+func NewLinearWatcherSource(svc LinearService, log *logger.Logger) *LinearWatcherSource {
+	return &LinearWatcherSource{service: svc, logger: log}
+}
+
+func (s *LinearWatcherSource) Name() string { return "linear" }
+
+func (s *LinearWatcherSource) Reserve(ctx context.Context, evt any) (bool, error) {
+	e, ok := evt.(*linear.NewLinearIssueEvent)
+	if !ok || e == nil || e.Issue == nil {
+		return false, errors.New("linear source: event payload missing or wrong type")
+	}
+	// Match today's reserveLinearIssue: nil service is "fail open".
+	if s.service == nil {
+		return true, nil
+	}
+	return s.service.ReserveIssueWatchTask(ctx, e.IssueWatchID, e.Issue.Identifier, e.Issue.URL)
+}
+
+func (s *LinearWatcherSource) Release(ctx context.Context, evt any) {
+	e, ok := evt.(*linear.NewLinearIssueEvent)
+	if !ok || e == nil || e.Issue == nil || s.service == nil {
+		return
+	}
+	if err := s.service.ReleaseIssueWatchTask(ctx, e.IssueWatchID, e.Issue.Identifier); err != nil && s.logger != nil {
+		s.logger.Warn("linear source: release failed",
+			zap.String("identifier", e.Issue.Identifier), zap.Error(err))
+	}
+}
+
+func (s *LinearWatcherSource) BuildTaskRequest(evt any) (*IssueTaskRequest, error) {
+	e, ok := evt.(*linear.NewLinearIssueEvent)
+	if !ok || e == nil || e.Issue == nil {
+		return nil, errors.New("linear source: event payload missing or wrong type")
+	}
+	return &IssueTaskRequest{
+		WorkspaceID:    e.WorkspaceID,
+		WorkflowID:     e.WorkflowID,
+		WorkflowStepID: e.WorkflowStepID,
+		Title:          fmt.Sprintf("[%s] %s", e.Issue.Identifier, e.Issue.Title),
+		Description:    interpolateLinearPrompt(e.Prompt, e.Issue),
+		Metadata: map[string]interface{}{
+			"linear_issue_watch_id":         e.IssueWatchID,
+			"linear_issue_identifier":       e.Issue.Identifier,
+			"linear_issue_url":              e.Issue.URL,
+			"linear_state":                  e.Issue.StateName,
+			"linear_assignee":               e.Issue.AssigneeName,
+			models.MetaKeyAgentProfileID:    e.AgentProfileID,
+			models.MetaKeyExecutorProfileID: e.ExecutorProfileID,
+		},
+	}, nil
+}
+
+func (s *LinearWatcherSource) AttachTaskID(ctx context.Context, evt any, taskID string) error {
+	e, ok := evt.(*linear.NewLinearIssueEvent)
+	if !ok || e == nil || e.Issue == nil || s.service == nil {
+		return nil
+	}
+	return s.service.AssignIssueWatchTaskID(ctx, e.IssueWatchID, e.Issue.Identifier, taskID)
+}
+
+func (s *LinearWatcherSource) AutoStartParams(evt any) AutoStartParams {
+	e, ok := evt.(*linear.NewLinearIssueEvent)
+	if !ok || e == nil {
+		return AutoStartParams{}
+	}
+	return AutoStartParams{
+		AgentProfileID:    e.AgentProfileID,
+		ExecutorProfileID: e.ExecutorProfileID,
+		Prompt:            interpolateLinearPrompt(e.Prompt, e.Issue),
+		WorkflowStepID:    e.WorkflowStepID,
+	}
+}

apps/backend/internal/orchestrator/source_linear_test.go

@@ -0,0 +1,178 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kandev/kandev/internal/linear"
+)
+
+type fakeLinearService struct {
+	reserveOK  bool
+	reserveErr error
+	assignErr  error
+	releaseErr error
+	gotReserve []string
+	gotAssign  []string
+	gotRelease []string
+}
+
+func (f *fakeLinearService) ReserveIssueWatchTask(_ context.Context, watchID, id, _ string) (bool, error) {
+	f.gotReserve = append(f.gotReserve, watchID+":"+id)
+	return f.reserveOK, f.reserveErr
+}
+
+func (f *fakeLinearService) AssignIssueWatchTaskID(_ context.Context, watchID, id, taskID string) error {
+	f.gotAssign = append(f.gotAssign, watchID+":"+id+":"+taskID)
+	return f.assignErr
+}
+
+func (f *fakeLinearService) ReleaseIssueWatchTask(_ context.Context, watchID, id string) error {
+	f.gotRelease = append(f.gotRelease, watchID+":"+id)
+	return f.releaseErr
+}
+
+func sampleLinearEvent() *linear.NewLinearIssueEvent {
+	return &linear.NewLinearIssueEvent{
+		IssueWatchID:      "watch-1",
+		WorkspaceID:       "ws-1",
+		WorkflowID:        "wf-1",
+		WorkflowStepID:    "step-1",
+		AgentProfileID:    "agent-1",
+		ExecutorProfileID: "exec-1",
+		Prompt:            "Work on {{issue.identifier}}: {{issue.title}}",
+		Issue: &linear.LinearIssue{
+			Identifier:    "ENG-1",
+			Title:         "Bug",
+			URL:           "https://linear.app/x/issue/ENG-1",
+			StateName:     "Todo",
+			AssigneeName:  "Alice",
+			CreatorName:   "Bob",
+			Description:   "details",
+			PriorityLabel: "High",
+			TeamKey:       "ENG",
+		},
+	}
+}
+
+func TestLinearSource_Name(t *testing.T) {
+	src := &LinearWatcherSource{}
+	if src.Name() != "linear" {
+		t.Fatalf("expected name=linear, got %q", src.Name())
+	}
+}
+
+func TestLinearSource_Reserve_Passthrough(t *testing.T) {
+	svc := &fakeLinearService{reserveOK: true}
+	src := &LinearWatcherSource{service: svc}
+	ok, err := src.Reserve(context.Background(), sampleLinearEvent())
+	if err != nil || !ok {
+		t.Fatalf("expected reserve ok, got ok=%v err=%v", ok, err)
+	}
+	if len(svc.gotReserve) != 1 || svc.gotReserve[0] != "watch-1:ENG-1" {
+		t.Fatalf("unexpected reserve args: %v", svc.gotReserve)
+	}
+}
+
+func TestLinearSource_Reserve_NilServiceFailOpen(t *testing.T) {
+	// Matches today's reserveLinearIssue: when service is unset (boot order
+	// corner case) the pipeline proceeds — better to risk a dup than drop.
+	src := &LinearWatcherSource{service: nil}
+	ok, err := src.Reserve(context.Background(), sampleLinearEvent())
+	if err != nil || !ok {
+		t.Fatalf("expected nil service to fail open, got ok=%v err=%v", ok, err)
+	}
+}
+
+func TestLinearSource_Reserve_Error(t *testing.T) {
+	svc := &fakeLinearService{reserveErr: errors.New("boom")}
+	src := &LinearWatcherSource{service: svc}
+	ok, err := src.Reserve(context.Background(), sampleLinearEvent())
+	if ok {
+		t.Fatal("expected reserve to fail")
+	}
+	if err == nil {
+		t.Fatal("expected reserve error to surface")
+	}
+}
+
+func TestLinearSource_BuildTaskRequest(t *testing.T) {
+	src := &LinearWatcherSource{}
+	req, err := src.BuildTaskRequest(sampleLinearEvent())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if req.Title != "[ENG-1] Bug" {
+		t.Errorf("title = %q", req.Title)
+	}
+	if req.WorkspaceID != "ws-1" || req.WorkflowID != "wf-1" || req.WorkflowStepID != "step-1" {
+		t.Errorf("workflow fields wrong: %+v", req)
+	}
+	if req.Description != "Work on ENG-1: Bug" {
+		t.Errorf("prompt interpolation wrong: %q", req.Description)
+	}
+	if req.Metadata["linear_issue_watch_id"] != "watch-1" {
+		t.Errorf("missing linear_issue_watch_id metadata")
+	}
+	if req.Metadata["linear_issue_identifier"] != "ENG-1" {
+		t.Errorf("missing linear_issue_identifier metadata")
+	}
+	if req.Metadata["agent_profile_id"] != "agent-1" {
+		t.Errorf("missing agent_profile_id metadata")
+	}
+}
+
+func TestLinearSource_BuildTaskRequest_WrongType(t *testing.T) {
+	src := &LinearWatcherSource{}
+	_, err := src.BuildTaskRequest("not an event")
+	if err == nil {
+		t.Fatal("expected error for wrong event type")
+	}
+}
+
+func TestLinearSource_AttachTaskID(t *testing.T) {
+	svc := &fakeLinearService{}
+	src := &LinearWatcherSource{service: svc}
+	if err := src.AttachTaskID(context.Background(), sampleLinearEvent(), "task-9"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(svc.gotAssign) != 1 || svc.gotAssign[0] != "watch-1:ENG-1:task-9" {
+		t.Fatalf("unexpected assign args: %v", svc.gotAssign)
+	}
+}
+
+func TestLinearSource_Release(t *testing.T) {
+	svc := &fakeLinearService{}
+	src := &LinearWatcherSource{service: svc}
+	src.Release(context.Background(), sampleLinearEvent())
+	if len(svc.gotRelease) != 1 || svc.gotRelease[0] != "watch-1:ENG-1" {
+		t.Fatalf("unexpected release args: %v", svc.gotRelease)
+	}
+}
+
+func TestLinearSource_Release_ErrorIsLoggedNotPropagated(t *testing.T) {
+	svc := &fakeLinearService{releaseErr: errors.New("dedup store down")}
+	src := &LinearWatcherSource{service: svc, logger: nopLogger(t)}
+	// Release returns no value: failure must be swallowed and logged.
+	// We only assert it does not panic and that the underlying service
+	// was still asked to release.
+	src.Release(context.Background(), sampleLinearEvent())
+	if len(svc.gotRelease) != 1 {
+		t.Fatalf("expected release call to be attempted, got %d", len(svc.gotRelease))
+	}
+}
+
+func TestLinearSource_AutoStartParams(t *testing.T) {
+	src := &LinearWatcherSource{}
+	p := src.AutoStartParams(sampleLinearEvent())
+	if p.AgentProfileID != "agent-1" || p.ExecutorProfileID != "exec-1" {
+		t.Fatalf("unexpected auto-start params: %+v", p)
+	}
+	if p.WorkflowStepID != "step-1" {
+		t.Errorf("step id wrong: %q", p.WorkflowStepID)
+	}
+	if p.Prompt != "Work on ENG-1: Bug" {
+		t.Errorf("prompt should be interpolated, got %q", p.Prompt)
+	}
+}

apps/backend/internal/orchestrator/watcher_dispatch.go

@@ -0,0 +1,146 @@
+package orchestrator
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"github.com/kandev/kandev/internal/common/logger"
+)
+
+// Log field keys repeated across coordinator log calls. Hoisted to file
+// constants so a typo would be a compile error rather than a silent log-
+// search miss.
+const (
+	logFieldSource = "source"
+	logFieldTaskID = "task_id"
+)
+
+// WatcherDispatchCoordinator owns the cross-integration pipeline that turns
+// a freshly-observed external issue (Linear, Jira, future: GitHub issues,
+// webhooks) into a Kandev task. It is the single seam where throttling,
+// observability, retry, or fairness will land — integration-specific code
+// stays in WatcherSource implementations.
+//
+// Pipeline:
+//
+//	Reserve → BuildTaskRequest → CreateIssueTask → AttachTaskID
+//	       → (optional) StartTask
+//
+// On any failure between Reserve and a successful CreateIssueTask, Release
+// is invoked so the dedup row does not strand the issue.
+type WatcherDispatchCoordinator struct {
+	taskCreator     IssueTaskCreator
+	startTask       taskStarter
+	shouldAutoStart func(ctx context.Context, workflowStepID string) bool
+	logger          *logger.Logger
+}
+
+// taskStarter wraps Service.StartTask so the coordinator can be tested
+// without spinning up the full orchestrator service.
+type taskStarter interface {
+	Start(ctx context.Context, taskID, workflowStepID string, params AutoStartParams) error
+}
+
+// AutoStartParams is the data a source contributes when the resulting task's
+// workflow step has auto-start enabled.
+type AutoStartParams struct {
+	AgentProfileID    string
+	ExecutorProfileID string
+	Prompt            string
+	WorkflowStepID    string
+}
+
+// WatcherSource encapsulates everything integration-specific about turning
+// a freshly-observed external issue into a Kandev task. Each method receives
+// the bus event payload as `any`; implementations type-assert at the top.
+// A failed assertion is a programming error (the subscriber wired the wrong
+// source) — implementations panic via the assertion's `ok` branch.
+type WatcherSource interface {
+	// Name returns a stable identifier ("linear", "jira", ...). Used for
+	// metrics labels and log fields.
+	Name() string
+
+	// Reserve atomically claims the dedup slot for this event. Returns
+	// (false, nil) when another concurrent reserver already won the race —
+	// the coordinator treats that as "nothing to do".
+	Reserve(ctx context.Context, evt any) (bool, error)
+
+	// Release rolls back a reservation when downstream work fails. Best
+	// effort; errors are logged but not surfaced.
+	Release(ctx context.Context, evt any)
+
+	// BuildTaskRequest translates the event into the shape the task creator
+	// expects. Returning an error triggers Release.
+	BuildTaskRequest(evt any) (*IssueTaskRequest, error)
+
+	// AttachTaskID writes the freshly-created task id back onto the dedup
+	// row so a future re-observation can short-circuit. Errors are logged
+	// but do not stop the pipeline — matching existing behaviour.
+	AttachTaskID(ctx context.Context, evt any, taskID string) error
+
+	// AutoStartParams returns the parameters needed to kick the task off
+	// when its workflow step is configured for auto-start.
+	AutoStartParams(evt any) AutoStartParams
+}
+
+// Dispatch runs one event through the full pipeline. Safe to call from a
+// goroutine; callers typically do so in the bus subscriber.
+func (c *WatcherDispatchCoordinator) Dispatch(ctx context.Context, src WatcherSource, evt any) {
+	reserved, err := src.Reserve(ctx, evt)
+	if err != nil {
+		c.logger.Error("watcher dispatch: reserve failed",
+			zap.String(logFieldSource, src.Name()), zap.Error(err))
+		return
+	}
+	if !reserved {
+		c.logger.Debug("watcher dispatch: already reserved by concurrent handler",
+			zap.String(logFieldSource, src.Name()))
+		return
+	}
+
+	req, err := src.BuildTaskRequest(evt)
+	if err != nil {
+		c.logger.Error("watcher dispatch: build task request failed",
+			zap.String(logFieldSource, src.Name()), zap.Error(err))
+		src.Release(ctx, evt)
+		return
+	}
+
+	task, err := c.taskCreator.CreateIssueTask(ctx, req)
+	if err != nil {
+		c.logger.Error("watcher dispatch: create issue task failed",
+			zap.String(logFieldSource, src.Name()), zap.Error(err))
+		src.Release(ctx, evt)
+		return
+	}
+
+	if err := src.AttachTaskID(ctx, evt, task.ID); err != nil {
+		c.logger.Error("watcher dispatch: attach task id failed",
+			zap.String(logFieldSource, src.Name()),
+			zap.String(logFieldTaskID, task.ID),
+			zap.Error(err))
+		// Do NOT release here — matches existing Linear/Jira behaviour:
+		// attach is a best-effort step, the task is already created.
+	}
+
+	c.logger.Info("watcher dispatch: created issue task",
+		zap.String(logFieldSource, src.Name()),
+		zap.String(logFieldTaskID, task.ID))
+
+	if !c.shouldAutoStart(ctx, req.WorkflowStepID) {
+		return
+	}
+
+	params := src.AutoStartParams(evt)
+	if err := c.startTask.Start(ctx, task.ID, req.WorkflowStepID, params); err != nil {
+		c.logger.Error("watcher dispatch: auto-start failed",
+			zap.String(logFieldSource, src.Name()),
+			zap.String(logFieldTaskID, task.ID),
+			zap.Error(err))
+		return
+	}
+	c.logger.Info("watcher dispatch: auto-started issue task",
+		zap.String(logFieldSource, src.Name()),
+		zap.String(logFieldTaskID, task.ID))
+}

apps/backend/internal/orchestrator/watcher_dispatch_test.go

@@ -0,0 +1,289 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"go.uber.org/zap"
+
+	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// nopLogger returns a *logger.Logger backed by zap.NewNop() for use in tests.
+// (The shared logger package does not expose a NewNop helper today.)
+func nopLogger(t *testing.T) *logger.Logger {
+	t.Helper()
+	l, err := logger.NewFromZap(zap.NewNop())
+	if err != nil {
+		t.Fatalf("failed to build nop logger: %v", err)
+	}
+	return l
+}
+
+// fakeWatcherSource is a configurable test double for WatcherSource. Tests
+// flip the public fields to drive coordinator branches; recorded* fields let
+// the test assert which coordinator stages actually ran.
+type fakeWatcherSource struct {
+	name             string
+	reserveOK        bool
+	reserveErr       error
+	buildReq         *IssueTaskRequest
+	buildErr         error
+	attachErr        error
+	autoStart        AutoStartParams
+	recordedReserve  int
+	recordedRelease  int
+	recordedBuild    int
+	recordedAttach   int
+	recordedAutoArgs *AutoStartParams
+}
+
+func (f *fakeWatcherSource) Name() string { return f.name }
+
+func (f *fakeWatcherSource) Reserve(_ context.Context, _ any) (bool, error) {
+	f.recordedReserve++
+	return f.reserveOK, f.reserveErr
+}
+
+func (f *fakeWatcherSource) Release(_ context.Context, _ any) {
+	f.recordedRelease++
+}
+
+func (f *fakeWatcherSource) BuildTaskRequest(_ any) (*IssueTaskRequest, error) {
+	f.recordedBuild++
+	return f.buildReq, f.buildErr
+}
+
+func (f *fakeWatcherSource) AttachTaskID(_ context.Context, _ any, _ string) error {
+	f.recordedAttach++
+	return f.attachErr
+}
+
+func (f *fakeWatcherSource) AutoStartParams(_ any) AutoStartParams {
+	p := f.autoStart
+	f.recordedAutoArgs = &p
+	return p
+}
+
+// fakeTaskCreator captures the request and returns a canned task or error.
+type fakeTaskCreator struct {
+	createErr error
+	gotReq    *IssueTaskRequest
+	returned  *models.Task
+}
+
+func (f *fakeTaskCreator) CreateIssueTask(_ context.Context, req *IssueTaskRequest) (*models.Task, error) {
+	f.gotReq = req
+	if f.createErr != nil {
+		return nil, f.createErr
+	}
+	if f.returned == nil {
+		f.returned = &models.Task{ID: "task-1"}
+	}
+	return f.returned, nil
+}
+
+func newTestCoordinator(t *testing.T, tc *fakeTaskCreator, autoStart bool, starter taskStarter) *WatcherDispatchCoordinator {
+	t.Helper()
+	return &WatcherDispatchCoordinator{
+		taskCreator: tc,
+		logger:      nopLogger(t),
+		shouldAutoStart: func(_ context.Context, _ string) bool {
+			return autoStart
+		},
+		startTask: starter,
+	}
+}
+
+// taskStarter is the indirection used inside Coordinator so tests can
+// inspect whether StartTask was invoked.
+type fakeTaskStarter struct {
+	called  int
+	gotID   string
+	gotStep string
+	err     error
+}
+
+func (f *fakeTaskStarter) Start(_ context.Context, taskID, stepID string, _ AutoStartParams) error {
+	f.called++
+	f.gotID = taskID
+	f.gotStep = stepID
+	return f.err
+}
+
+func TestCoordinator_Dispatch_HappyPath(t *testing.T) {
+	src := &fakeWatcherSource{
+		name:      "linear",
+		reserveOK: true,
+		buildReq: &IssueTaskRequest{
+			WorkspaceID:    "ws-1",
+			WorkflowID:     "wf-1",
+			WorkflowStepID: "step-1",
+			Title:          "[ENG-1] Hello",
+			Description:    "body",
+		},
+		autoStart: AutoStartParams{
+			AgentProfileID:    "agent-1",
+			ExecutorProfileID: "exec-1",
+			Prompt:            "body",
+			WorkflowStepID:    "step-1",
+		},
+	}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedReserve != 1 || src.recordedBuild != 1 || src.recordedAttach != 1 {
+		t.Fatalf("expected reserve/build/attach each once, got R=%d B=%d A=%d",
+			src.recordedReserve, src.recordedBuild, src.recordedAttach)
+	}
+	if src.recordedRelease != 0 {
+		t.Fatalf("expected no release on happy path, got %d", src.recordedRelease)
+	}
+	if tc.gotReq == nil || tc.gotReq.Title != "[ENG-1] Hello" {
+		t.Fatalf("expected task creator to receive built request, got %+v", tc.gotReq)
+	}
+	if starter.called != 1 {
+		t.Fatalf("expected auto-start to fire once, got %d", starter.called)
+	}
+	if starter.gotID != "task-1" || starter.gotStep != "step-1" {
+		t.Fatalf("auto-start received wrong args: id=%s step=%s", starter.gotID, starter.gotStep)
+	}
+}
+
+func TestCoordinator_Dispatch_ReserveLost(t *testing.T) {
+	src := &fakeWatcherSource{name: "linear", reserveOK: false}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedReserve != 1 {
+		t.Fatalf("expected reserve to be attempted, got %d", src.recordedReserve)
+	}
+	if src.recordedBuild != 0 || src.recordedAttach != 0 || starter.called != 0 || tc.gotReq != nil {
+		t.Fatalf("expected pipeline to stop after reserve=false, got B=%d A=%d Start=%d gotReq=%v",
+			src.recordedBuild, src.recordedAttach, starter.called, tc.gotReq)
+	}
+}
+
+func TestCoordinator_Dispatch_ReserveError(t *testing.T) {
+	src := &fakeWatcherSource{name: "linear", reserveOK: false, reserveErr: errors.New("db down")}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedBuild != 0 || tc.gotReq != nil {
+		t.Fatal("expected pipeline to stop after reserve error")
+	}
+}
+
+func TestCoordinator_Dispatch_BuildError_Releases(t *testing.T) {
+	src := &fakeWatcherSource{name: "linear", reserveOK: true, buildErr: errors.New("bad payload")}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedRelease != 1 {
+		t.Fatalf("expected release after build error, got %d", src.recordedRelease)
+	}
+	if tc.gotReq != nil || starter.called != 0 {
+		t.Fatal("expected pipeline to stop after build error")
+	}
+}
+
+func TestCoordinator_Dispatch_CreateError_Releases(t *testing.T) {
+	src := &fakeWatcherSource{
+		name:      "linear",
+		reserveOK: true,
+		buildReq:  &IssueTaskRequest{WorkspaceID: "ws-1"},
+	}
+	tc := &fakeTaskCreator{createErr: errors.New("write failed")}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedRelease != 1 {
+		t.Fatalf("expected release after create error, got %d", src.recordedRelease)
+	}
+	if src.recordedAttach != 0 || starter.called != 0 {
+		t.Fatal("expected pipeline to stop after create error")
+	}
+}
+
+func TestCoordinator_Dispatch_NoAutoStart(t *testing.T) {
+	src := &fakeWatcherSource{
+		name:      "linear",
+		reserveOK: true,
+		buildReq:  &IssueTaskRequest{WorkspaceID: "ws-1", WorkflowStepID: "step-1"},
+	}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, false, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if src.recordedAttach != 1 {
+		t.Fatalf("expected attach to run, got %d", src.recordedAttach)
+	}
+	if starter.called != 0 {
+		t.Fatalf("expected no auto-start when shouldAutoStart=false, got %d", starter.called)
+	}
+}
+
+func TestCoordinator_Dispatch_AttachError_LogsButContinues(t *testing.T) {
+	src := &fakeWatcherSource{
+		name:      "linear",
+		reserveOK: true,
+		buildReq:  &IssueTaskRequest{WorkspaceID: "ws-1", WorkflowStepID: "step-1"},
+		attachErr: errors.New("attach failed"),
+	}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	// Today's createLinearIssueTask logs the attach failure but still auto-starts;
+	// preserve that behaviour exactly.
+	if starter.called != 1 {
+		t.Fatalf("expected auto-start to still run after attach error, got %d", starter.called)
+	}
+	// Attach failure is best-effort: the task was already created, so we
+	// MUST NOT Release the dedup row — that would orphan the task.
+	if src.recordedRelease != 0 {
+		t.Fatalf("expected no Release after attach error, got %d", src.recordedRelease)
+	}
+}
+
+func TestCoordinator_Dispatch_AutoStartError_LogsAndReturns(t *testing.T) {
+	src := &fakeWatcherSource{
+		name:      "linear",
+		reserveOK: true,
+		buildReq:  &IssueTaskRequest{WorkspaceID: "ws-1", WorkflowStepID: "step-1"},
+	}
+	tc := &fakeTaskCreator{}
+	starter := &fakeTaskStarter{err: errors.New("start failed")}
+	c := newTestCoordinator(t, tc, true, starter)
+
+	c.Dispatch(context.Background(), src, "evt")
+
+	if starter.called != 1 {
+		t.Fatalf("expected auto-start to be attempted, got %d", starter.called)
+	}
+	// Auto-start failure is terminal: the task exists and the dedup row is
+	// already attached. We must not Release.
+	if src.recordedRelease != 0 {
+		t.Fatalf("expected no Release after auto-start error, got %d", src.recordedRelease)
+	}
+}

apps/backend/internal/orchestrator/watcher_dispatch_wiring.go

@@ -0,0 +1,69 @@
+package orchestrator
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+)
+
+// serviceTaskStarter adapts Service.StartTask to the coordinator's
+// taskStarter interface. Lives in its own file so the wiring stays close
+// to the Service definition without polluting watcher_dispatch.go with
+// orchestrator-internal types.
+type serviceTaskStarter struct{ svc *Service }
+
+func (s serviceTaskStarter) Start(ctx context.Context, taskID, workflowStepID string, p AutoStartParams) error {
+	_, err := s.svc.StartTask(
+		ctx, taskID, p.AgentProfileID, "", p.ExecutorProfileID,
+		"", p.Prompt, workflowStepID, false, true, nil,
+	)
+	return err
+}
+
+// initWatcherCoordinator builds the coordinator (once) and (always) refreshes
+// the mutable taskCreator dependency. Called from SetIssueTaskCreator, which
+// can be invoked multiple times — tests in particular may swap creators
+// between scenarios. Re-running the setter MUST update the coordinator,
+// otherwise Dispatch silently keeps the original creator.
+func (s *Service) initWatcherCoordinator() {
+	if s.watcherCoordinator == nil {
+		s.watcherCoordinator = &WatcherDispatchCoordinator{
+			startTask: serviceTaskStarter{svc: s},
+			shouldAutoStart: func(ctx context.Context, stepID string) bool {
+				return s.shouldAutoStartStep(ctx, stepID)
+			},
+			logger: s.logger,
+		}
+	}
+	// Always refresh: SetIssueTaskCreator may be called more than once.
+	s.watcherCoordinator.taskCreator = s.issueTaskCreator
+}
+
+// dispatchWatcherEvent runs the wiring guards every per-integration bus
+// handler shares — issueTaskCreator check, coordinator check, and the
+// final goroutine dispatch with cancellation detached. integration is used
+// in log message templating ("new linear issue ...", "skipping jira task ...").
+// fields are the structured log fields that identify the event in operator
+// logs; pass at least the issue_watch_id and an integration-specific
+// identifier field so a deferred / dropped event is diagnosable.
+//
+// Lives in its own helper so per-integration handlers stay below dupl's
+// duplicate-block threshold without copy-pasting the same guards.
+func (s *Service) dispatchWatcherEvent(ctx context.Context, integration string, src WatcherSource, evt any, fields ...zap.Field) {
+	s.logger.Info(fmt.Sprintf("new %s issue detected from watch", integration), fields...)
+	if s.issueTaskCreator == nil {
+		s.logger.Warn(fmt.Sprintf("issue task creator not configured, skipping %s task creation", integration))
+		return
+	}
+	if s.watcherCoordinator == nil {
+		// Defensive: coordinator is wired by SetIssueTaskCreator. If we got
+		// here without the creator we already returned above; this is just
+		// a belt-and-braces guard for tests that wire pieces individually.
+		s.logger.Warn(fmt.Sprintf("watcher coordinator not configured, skipping %s task dispatch", integration), fields...)
+		return
+	}
+	// Detach from cancellation but keep request-scoped values (tracing, etc.):
+	// the bus delivery context may be cancelled before task creation finishes.
+	go s.watcherCoordinator.Dispatch(context.WithoutCancel(ctx), src, evt)
+}