diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 6bfd670629..cecff267be 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -1470,6 +1470,10 @@ func wireOfficeSvcsDependencies( // Route the Office "No parent" mutation through the canonical task detach // operation so inherited workspace sharing remains valid. services.OfficeSvcs.Dashboard.SetTaskDetacher(services.Task) + // Publish task.updated for Office status changes so WS-driven UI outside + // the Office board (All-Workflows kanban, task views, task/statussummary) + // sees the mutation instead of rendering stale state until a refetch. + services.OfficeSvcs.Dashboard.SetTaskLifecyclePublisher(services.Task) // Human assignee writes from the office PATCH surface go through the task // service, which authorizes the caller and validates the assignee. That // route carries no :wsId, so it is not covered by the office diff --git a/apps/backend/internal/office/dashboard/service.go b/apps/backend/internal/office/dashboard/service.go index 2c20f8658f..53aafc317e 100644 --- a/apps/backend/internal/office/dashboard/service.go +++ b/apps/backend/internal/office/dashboard/service.go @@ -191,6 +191,14 @@ type TaskDetacher interface { DetachTask(ctx context.Context, taskID string) (*taskmodels.Task, error) } +// TaskLifecyclePublisher reloads and publishes the canonical task.updated +// event for a task row Office has just mutated. The implementation owns task +// publication ordering so concurrent status writes cannot publish stale +// snapshots. Office's own status-change events only reach the Office board. +type TaskLifecyclePublisher interface { + PublishTaskUpdatedByID(ctx context.Context, id string) +} + // SessionTerminator flips the (task, agent) office session row to a terminal // state. Used when an agent stops being a participant on a task — reassignment, // reviewer/approver removal, or agent instance deletion. Idempotent: skipping @@ -378,6 +386,7 @@ type DashboardService struct { retryCanceller RetryCanceller // optional; nil means retries are not cancelled on reassign taskCanceller TaskCanceller // optional; used to hard-cancel sessions on status→cancelled taskDetacher TaskDetacher // optional; canonical empty-parent mutation + taskLifecycle TaskLifecyclePublisher // optional; nil means status changes don't publish canonical task.updated sessionTerm SessionTerminator // optional; flips office session rows to COMPLETED on participation removal reactivity ReactivityApplier // optional; runs the office reactivity pipeline on mutations engineDispatcher shared.WorkflowEngineDispatcher // optional; synchronously routes comment triggers through the engine @@ -569,6 +578,12 @@ func (s *DashboardService) SetTaskDetacher(d TaskDetacher) { s.taskDetacher = d } +// SetTaskLifecyclePublisher wires the canonical task.updated publisher used +// after a status change persists a task row mutation. +func (s *DashboardService) SetTaskLifecyclePublisher(p TaskLifecyclePublisher) { + s.taskLifecycle = p +} + // SetSessionTerminator wires the office session terminator. Optional; when // nil, participation removal won't flip the prior agent's session to a // terminal state (the row stays IDLE/RUNNING and its conversation lingers). diff --git a/apps/backend/internal/office/dashboard/service_tasks.go b/apps/backend/internal/office/dashboard/service_tasks.go index 64b0837cc1..b83b974bda 100644 --- a/apps/backend/internal/office/dashboard/service_tasks.go +++ b/apps/backend/internal/office/dashboard/service_tasks.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/kandev/kandev/internal/common/taskdependencies" "github.com/kandev/kandev/internal/events" @@ -628,6 +629,7 @@ func (s *DashboardService) UpdateTaskStatus(ctx context.Context, req TaskStatusU commentID := s.maybeCreateStatusComment(ctx, req) s.logTaskStatusChangeActivity(ctx, req) s.publishTaskStatusChanged(ctx, req) + s.publishCanonicalTaskUpdated(ctx, req.TaskID) s.runReactivityForStatus(ctx, req, commentID, preStatus) s.maybeSupersedeOnRework(ctx, req.TaskID, preStatus, dbState) @@ -984,6 +986,29 @@ func (s *DashboardService) publishTaskStatusChanged(ctx context.Context, req Tas } } +// canonicalTaskUpdatedPublishTimeout bounds the detached reload+publish in +// publishCanonicalTaskUpdated so a caller-cancelled ctx can't hang it forever. +const canonicalTaskUpdatedPublishTimeout = 10 * time.Second + +// publishCanonicalTaskUpdated publishes the canonical task.updated event for +// a task row this function has just mutated via s.repo.UpdateTaskState. +// office.task.status_changed above only reaches the Office board; +// task.updated is what WS-driven UI outside Office (the All-Workflows +// kanban view, task views, the task/statussummary projector) keys off. +// Nil-safe: skipped when no publisher is wired. Runs on a context detached +// from ctx's cancellation: the mutation has already committed, so a caller +// that disconnects (HTTP) or a ctx that expires after the write must not +// suppress the event other WS-driven views depend on. The task service owns +// the reload and per-task publication queue. +func (s *DashboardService) publishCanonicalTaskUpdated(ctx context.Context, taskID string) { + if s.taskLifecycle == nil { + return + } + pubCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), canonicalTaskUpdatedPublishTimeout) + defer cancel() + s.taskLifecycle.PublishTaskUpdatedByID(pubCtx, taskID) +} + // runReactivityForComment fires the pipeline for a standalone comment // (one not attached to a status update). Wakes the assignee unless the // comment is a self-comment or the task is closed; resolves @mentions. diff --git a/apps/backend/internal/office/dashboard/task_updated_cancellation_internal_test.go b/apps/backend/internal/office/dashboard/task_updated_cancellation_internal_test.go new file mode 100644 index 0000000000..9170077ac2 --- /dev/null +++ b/apps/backend/internal/office/dashboard/task_updated_cancellation_internal_test.go @@ -0,0 +1,47 @@ +package dashboard + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/common/logger" +) + +// ctxCapturingPublisher is a fake TaskLifecyclePublisher that records the +// ctx.Err() observed by each call, so a test can assert the ctx actually +// used to reach the event bus is not the caller's already-cancelled one. +type ctxCapturingPublisher struct { + publishCalled bool + publishCtxErr error + publishedID string +} + +func (p *ctxCapturingPublisher) PublishTaskUpdatedByID(ctx context.Context, id string) { + p.publishCalled = true + p.publishCtxErr = ctx.Err() + p.publishedID = id +} + +// TestPublishCanonicalTaskUpdated_SurvivesCallerCancellation covers the +// window a caller (most commonly an HTTP request context) can be cancelled +// after UpdateTaskStatus's DB write commits: a request disconnect must not +// suppress the task.updated event other WS-driven views depend on. +func TestPublishCanonicalTaskUpdated_SurvivesCallerCancellation(t *testing.T) { + pub := &ctxCapturingPublisher{} + s := &DashboardService{taskLifecycle: pub, logger: logger.Default()} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // simulates the caller disconnecting right after the DB write commits + + s.publishCanonicalTaskUpdated(ctx, "task-1") + + if !pub.publishCalled { + t.Fatal("PublishTaskUpdatedByID was not called after the caller's context was cancelled") + } + if pub.publishCtxErr != nil { + t.Fatalf("PublishTaskUpdatedByID context was cancelled (err=%v), want a detached context", pub.publishCtxErr) + } + if pub.publishedID != "task-1" { + t.Fatalf("published task ID = %q, want task-1", pub.publishedID) + } +} diff --git a/apps/backend/internal/office/dashboard/task_updated_events_test.go b/apps/backend/internal/office/dashboard/task_updated_events_test.go new file mode 100644 index 0000000000..dff1605a73 --- /dev/null +++ b/apps/backend/internal/office/dashboard/task_updated_events_test.go @@ -0,0 +1,95 @@ +package dashboard_test + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/office/dashboard" + "github.com/kandev/kandev/internal/office/models" +) + +// recordingTaskLifecyclePublisher is a fake dashboard.TaskLifecyclePublisher +// that records every PublishTaskUpdatedByID call so tests can assert exactly +// one canonical task.updated reaches the bus per UpdateTaskStatus call. +type recordingTaskLifecyclePublisher struct { + published []string +} + +func (r *recordingTaskLifecyclePublisher) PublishTaskUpdatedByID(_ context.Context, id string) { + r.published = append(r.published, id) +} + +// TestUpdateTaskStatus_PublishesCanonicalTaskUpdated pins AGENTS.md:228: +// any code path that mutates a task row must publish task.created / +// task.updated / task.deleted. UpdateTaskStatus writes the tasks row via +// s.repo.UpdateTaskState directly and, before this fix, published only the +// Office-native office.task.status_changed event — invisible to WS-driven +// UI like the All-Workflows kanban view. +func TestUpdateTaskStatus_PublishesCanonicalTaskUpdated(t *testing.T) { + deps := newTestDeps(t) + insertTestTask(t, deps.db, "cu1", "ws-cu", "CU", "todo", 0) + pub := &recordingTaskLifecyclePublisher{} + deps.svc.SetTaskLifecyclePublisher(pub) + + if err := deps.svc.UpdateTaskStatus(context.Background(), dashboard.TaskStatusUpdateRequest{ + TaskID: "cu1", + NewStatus: "in_progress", + }); err != nil { + t.Fatalf("update task status: %v", err) + } + + if len(pub.published) != 1 || pub.published[0] != "cu1" { + t.Fatalf("published = %v, want exactly one task.updated for cu1", pub.published) + } +} + +// TestUpdateTaskStatus_PublishesCanonicalTaskUpdatedWhenGateRedirects is the +// card's named DoD assertion: a gate-redirected status change (approver +// pending) still persists a state change (COMPLETED -> REVIEW redirect) and +// must still publish exactly one task.updated, even though the call itself +// returns a typed *ApprovalsPendingError. +func TestUpdateTaskStatus_PublishesCanonicalTaskUpdatedWhenGateRedirects(t *testing.T) { + deps := newTestDeps(t) + insertTestTask(t, deps.db, "cu2", "ws-cu", "CU2", "in_progress", 0) + mustAddParticipant(t, deps, "cu2", "agent-A", models.ParticipantRoleApprover) + pub := &recordingTaskLifecyclePublisher{} + deps.svc.SetTaskLifecyclePublisher(pub) + + err := deps.svc.UpdateTaskStatus(context.Background(), dashboard.TaskStatusUpdateRequest{ + TaskID: "cu2", + NewStatus: "done", + }) + var pendingErr *dashboard.ApprovalsPendingError + if !errors.As(err, &pendingErr) { + t.Fatalf("err = %v, want *dashboard.ApprovalsPendingError", err) + } + + if len(pub.published) != 1 || pub.published[0] != "cu2" { + t.Fatalf("published = %v, want exactly one task.updated for cu2 despite the gate redirect", pub.published) + } + + exec, err := deps.repo.GetTaskExecutionFields(context.Background(), "cu2") + if err != nil { + t.Fatalf("get task execution fields: %v", err) + } + if exec == nil || exec.State != "REVIEW" { + t.Fatalf("persisted state = %+v, want REVIEW", exec) + } +} + +// TestUpdateTaskStatus_NoPublisherWiredIsSafe pins the nil-safe fallback: +// most existing DashboardService test harnesses (and any production +// deployment sequenced before main.go's wiring runs) never call +// SetTaskLifecyclePublisher, and UpdateTaskStatus must not panic. +func TestUpdateTaskStatus_NoPublisherWiredIsSafe(t *testing.T) { + deps := newTestDeps(t) + insertTestTask(t, deps.db, "cu3", "ws-cu", "CU3", "todo", 0) + + if err := deps.svc.UpdateTaskStatus(context.Background(), dashboard.TaskStatusUpdateRequest{ + TaskID: "cu3", + NewStatus: "in_progress", + }); err != nil { + t.Fatalf("update task status: %v", err) + } +} diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index 3bb6b8e208..764c0c075b 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -52,6 +52,27 @@ func (s *Service) PublishTaskUpdated(ctx context.Context, task *models.Task, old s.publishTaskEvent(ctx, events.TaskUpdated, task, nil, oldWorkflowIDs...) } +// PublishTaskUpdatedByID reloads and publishes the canonical task.updated +// event inside the task's FIFO publication queue. Callers use this after a +// direct task-row mutation when the database row is the source of truth. +func (s *Service) PublishTaskUpdatedByID(ctx context.Context, taskID string) { + if s.eventBus == nil || taskID == "" { + return + } + s.enqueueTaskPublication(ctx, taskID, events.TaskUpdated, func(publicationCtx context.Context) { + task, err := s.GetTask(publicationCtx, taskID) + if err != nil { + s.logger.Error("failed to load task for task.updated publication", + zap.String("task_id", taskID), zap.Error(err)) + return + } + if task == nil { + return + } + s.publishTaskEventNow(publicationCtx, events.TaskUpdated, task, nil, nil, nil, nil) + }) +} + // PublishTaskQueuePromoted notifies subscribers that a queued task has been // admitted to its destination step. The event is distinct from task.updated so // orchestration can launch deferred work exactly when capacity is granted. diff --git a/apps/backend/internal/task/service/service_events_test.go b/apps/backend/internal/task/service/service_events_test.go index 216cb23f89..7145d5b9a2 100644 --- a/apps/backend/internal/task/service/service_events_test.go +++ b/apps/backend/internal/task/service/service_events_test.go @@ -122,6 +122,83 @@ func TestTaskPublication_ActivityRefreshDoesNotOvertakeOrdinaryUpdate(t *testing <-activityDone } +type firstTaskReadBarrierRepository struct { + repository.TaskRepository + firstReadEntered chan struct{} + releaseFirstRead chan struct{} + + mu sync.Mutex + getCalls int +} + +func (r *firstTaskReadBarrierRepository) GetTask(ctx context.Context, id string) (*models.Task, error) { + task, err := r.TaskRepository.GetTask(ctx, id) + + r.mu.Lock() + r.getCalls++ + firstRead := r.getCalls == 1 + r.mu.Unlock() + if firstRead { + close(r.firstReadEntered) + <-r.releaseFirstRead + } + return task, err +} + +func TestPublishTaskUpdatedByID_ReloadsInsideTaskPublicationQueue(t *testing.T) { + svc, eventBus, repo := createTestService(t) + ctx := context.Background() + createTaskWithoutRepositories(t, ctx, repo) + if err := repo.UpdateTaskState(ctx, "task-1", v1.TaskStateTODO); err != nil { + t.Fatalf("initialize task state: %v", err) + } + + barrier := &firstTaskReadBarrierRepository{ + TaskRepository: svc.tasks, + firstReadEntered: make(chan struct{}), + releaseFirstRead: make(chan struct{}), + } + svc.tasks = barrier + + firstDone := make(chan struct{}) + go func() { + svc.PublishTaskUpdatedByID(ctx, "task-1") + close(firstDone) + }() + <-barrier.firstReadEntered + + if err := repo.UpdateTaskState(ctx, "task-1", v1.TaskStateInProgress); err != nil { + t.Fatalf("update task state: %v", err) + } + svc.PublishTaskUpdatedByID(ctx, "task-1") + + close(barrier.releaseFirstRead) + <-firstDone + + published := eventBus.GetPublishedEvents() + if len(published) != 2 { + t.Fatalf("published %d events, want 2", len(published)) + } + states := make([]string, 0, len(published)) + for _, event := range published { + data, ok := event.Data.(map[string]interface{}) + if !ok { + t.Fatalf("event data type = %T, want map[string]interface{}", event.Data) + } + state, ok := data["state"].(string) + if !ok { + t.Fatalf("event state type = %T, want string", data["state"]) + } + states = append(states, state) + } + if got, want := states[0], string(v1.TaskStateTODO); got != want { + t.Fatalf("first published state = %q, want %q", got, want) + } + if got, want := states[1], string(v1.TaskStateInProgress); got != want { + t.Fatalf("last published state = %q, want %q", got, want) + } +} + func TestTaskPublication_PreservesSameSecondActivityOrdering(t *testing.T) { svc, eventBus, _ := createTestService(t) messageAt := time.Date(2026, 8, 18, 10, 20, 0, 100_000_000, time.UTC)