fix(office): don't leave task views stale after an approval-gate redirect (#3492)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-10 21:50:00 +08:00
committed by GitHub
parent f26d8281d6
commit be19edfa4d
7 changed files with 284 additions and 0 deletions
+4
View File
@@ -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
@@ -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).
@@ -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.
@@ -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)
}
}
@@ -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)
}
}
@@ -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.
@@ -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)