fix(office): populate Started/Completed timestamps on task detail (#2856)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
nova28
2026-08-21 03:29:51 +08:00
committed by GitHub
parent fc663a9bba
commit 0243663ec6
32 changed files with 1071 additions and 244 deletions
+9
View File
@@ -1277,6 +1277,15 @@ func wireOfficeSvcsDependencies(
// relayed-comment activity rows get tagged with the originating
// run id (Tasks Touched on the run detail page).
services.OfficeSvcs.Channels.SetRunResolver(services.Office)
// Wire the office service as the dashboard's run resolver so the
// synchronous task_status_changed activity write (UpdateTaskStatus)
// tags its row with the originating run id, matching the async
// subscriber it replaced.
services.OfficeSvcs.Dashboard.SetRunResolver(services.Office)
// Wire the Office activity projection before task.state_changed events
// reach the WebSocket broadcaster, so workflow moves have durable timeline
// data when the frontend refetches the task detail.
services.Task.SetTaskStateActivityLogger(services.OfficeSvcs.Dashboard)
// Wire the office service as the retry canceller for task reassignment.
services.OfficeSvcs.Dashboard.SetRetryCanceller(services.Office)
// Wire the office service as the task canceller for status→cancelled hard-cancels.
@@ -0,0 +1,73 @@
package dashboard_test
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/jmoiron/sqlx"
"github.com/kandev/kandev/internal/office/dashboard"
)
// seedStatusChangeActivity inserts an office_activity_log row shaped like
// the one DashboardService.UpdateTaskStatus's real event-subscriber
// (handleTaskStatusChanged) would write in production. This test suite's
// DashboardService is built without an event bus (see newTestDeps), so
// calling UpdateTaskStatus directly here would not itself produce any
// activity rows; seed them explicitly instead, matching the pattern used
// by seedSummaryActivity in agent_summary_test.go.
func seedStatusChangeActivity(t *testing.T, db *sqlx.DB, taskID, wsID, newStatus string, at time.Time) {
t.Helper()
_, err := db.Exec(`
INSERT INTO office_activity_log
(id, workspace_id, actor_type, actor_id, action,
target_type, target_id, details, run_id, session_id, created_at)
VALUES (?, ?, 'system', 'office-scheduler', 'task_status_changed', 'task', ?, ?, '', '', ?)
`, "act-"+taskID+"-"+at.Format("20060102150405"), wsID, taskID,
fmt.Sprintf(`{"new_status":%q}`, newStatus), at.UTC().Format(time.RFC3339))
if err != nil {
t.Fatalf("seed status-change activity for %s: %v", taskID, err)
}
}
// TestGetTask_CompletedAtClearsWhenTaskReopened seeds a
// todo -> in_progress -> done -> in_progress activity history (the shape
// the office direct-update path writes) for a task currently in_progress,
// and asserts the sidebar no longer reports a completedAt once the task
// is reopened, even though it once completed. startedAt must survive the
// reopen unaffected.
func TestGetTask_CompletedAtClearsWhenTaskReopened(t *testing.T) {
deps := newTestDeps(t)
insertTestTask(t, deps.db, "reopen-task", "ws-reopen", "Reopen Task", "in_progress", 0)
base := time.Date(2026, 8, 1, 10, 0, 0, 0, time.UTC)
seedStatusChangeActivity(t, deps.db, "reopen-task", "ws-reopen", "in_progress", base)
seedStatusChangeActivity(t, deps.db, "reopen-task", "ws-reopen", "done", base.Add(time.Hour))
seedStatusChangeActivity(t, deps.db, "reopen-task", "ws-reopen", "in_progress", base.Add(2*time.Hour))
req := httptest.NewRequest(http.MethodGet, "/api/v1/office/tasks/reopen-task", nil)
w := httptest.NewRecorder()
deps.router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var resp dashboard.TaskResponse
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
t.Fatalf("decode: %v", err)
}
if resp.Task == nil {
t.Fatal("expected task in response")
}
if resp.Task.StartedAt == "" {
t.Error("expected startedAt to survive the reopen")
}
if resp.Task.CompletedAt != "" {
t.Errorf("expected completedAt cleared on a currently-open task, got %q", resp.Task.CompletedAt)
}
}
+10 -2
View File
@@ -376,8 +376,16 @@ type TaskDTO struct {
Reviewers []string `json:"reviewers"`
Approvers []string `json:"approvers"`
Decisions []*DecisionDTO `json:"decisions,omitempty"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
// StartedAt/CompletedAt are derived from the task's status-change
// activity log, not a persisted column (there is no started_at /
// completed_at on the tasks table). Only the detail handler
// (taskRowToDTO is called for both list and detail responses, but
// the derivation needs the timeline it doesn't have) sets these; see
// deriveTaskTimestamps in handler.go.
StartedAt string `json:"startedAt,omitempty"`
CompletedAt string `json:"completedAt,omitempty"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
// IsSystem flags tasks that live in a kandev-managed system
// workflow (today: standing coordination; future: routine-fired).
// The Office Tasks UI hides these by default and surfaces a small
@@ -462,6 +462,10 @@ func (h *Handler) getTask(c *gin.Context) {
h.attachDecisions(ctx, c, task.ID, dto)
statusChanges, _ := h.svc.ListStatusChanges(ctx, task.WorkspaceID, task.ID)
dto.StartedAt, dto.CompletedAt = deriveTaskTimestamps(statusChanges)
if !timelineStatusIsDone(dto.Status) {
dto.CompletedAt = ""
}
c.JSON(http.StatusOK, TaskResponse{Task: dto, Timeline: buildStatusTimeline(statusChanges)})
}
@@ -496,19 +500,6 @@ func (h *Handler) attachDecisions(ctx context.Context, c *gin.Context, taskID st
}
}
func buildStatusTimeline(changes []TimelineEvent) []TimelineEventDTO {
timeline := make([]TimelineEventDTO, len(changes))
for i, ev := range changes {
timeline[i] = TimelineEventDTO{
Type: "status_change",
From: ev.From,
To: ev.To,
At: ev.At,
}
}
return timeline
}
func taskRowToDTO(r *sqlite.TaskRow, lbls []*sqlite.Label) *TaskDTO {
labels := make([]LabelDTO, len(lbls))
for i, l := range lbls {
@@ -14,6 +14,7 @@ import (
"github.com/kandev/kandev/internal/office/shared"
taskmodels "github.com/kandev/kandev/internal/task/models"
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
"go.uber.org/zap"
)
@@ -144,6 +145,14 @@ type RetryCanceller interface {
CancelPendingRetriesForTask(ctx context.Context, taskID string) error
}
// RunResolver resolves the originating office run id for a task, used
// to attribute the task_status_changed activity row back to the run
// that produced the transition. Optional dependency — when nil, the
// activity row is logged with an empty run_id. Mirrors channels.RunResolver.
type RunResolver interface {
ResolveRunForTask(ctx context.Context, taskID string) string
}
// TaskCanceller hard-cancels a task's active execution. Used by the
// reactivity pipeline when a task is moved to "cancelled" status.
type TaskCanceller interface {
@@ -351,6 +360,7 @@ type DashboardService struct {
decisions DecisionStore // workflow-domain decisions store (ADR 0005 Wave E); nil disables decision endpoints
routingProvider RoutingProvider // optional; nil disables /routing endpoints (503)
attemptLister RouteAttemptLister // optional; nil disables attempt embedding on run-detail responses
runResolver RunResolver // optional; nil means status-change activity rows have no run_id
}
// SetRoutingProvider wires the provider-routing seam used by the
@@ -451,6 +461,43 @@ func (s *DashboardService) SetRetryCanceller(c RetryCanceller) {
s.retryCanceller = c
}
// SetRunResolver wires the seam used to attribute status-change activity
// rows back to the office run that produced them. Optional; when unset,
// rows are logged with an empty run_id.
func (s *DashboardService) SetRunResolver(r RunResolver) {
s.runResolver = r
}
// LogTaskStateChange records a task state transition for Office tasks before
// the task service publishes its state-change notification. This keeps the
// activity-backed timeline durable before clients refetch the task detail.
// Non-Office tasks use the shared task service and are ignored here.
func (s *DashboardService) LogTaskStateChange(
ctx context.Context, task *taskmodels.Task, oldState v1.TaskState,
) {
if task == nil || oldState == task.State || s.activity == nil {
return
}
fields, err := s.repo.GetTaskExecutionFields(ctx, task.ID)
if err != nil || fields == nil || !fields.IsFromOffice {
return
}
wsID := fields.WorkspaceID
if wsID == "" {
wsID = task.WorkspaceID
}
runID := ""
if s.runResolver != nil {
runID = s.runResolver.ResolveRunForTask(ctx, task.ID)
}
details, _ := json.Marshal(map[string]string{
"new_status": string(task.State),
"old_status": string(oldState),
})
s.activity.LogActivityWithRun(ctx, wsID, "system", "orchestrator",
"task_status_changed", "task", task.ID, string(details), runID, "")
}
// SetTaskCanceller sets the canceller used to hard-cancel sessions when
// a task is moved to "cancelled".
func (s *DashboardService) SetTaskCanceller(c TaskCanceller) {
@@ -476,6 +476,7 @@ func (s *DashboardService) UpdateTaskStatus(ctx context.Context, req TaskStatusU
}
commentID := s.maybeCreateStatusComment(ctx, req)
s.logTaskStatusChangeActivity(ctx, req)
s.publishTaskStatusChanged(ctx, req)
s.runReactivityForStatus(ctx, req, commentID, preStatus)
s.maybeSupersedeOnRework(ctx, req.TaskID, preStatus, dbState)
@@ -497,6 +498,38 @@ func (s *DashboardService) UpdateTaskStatusAsAgent(
})
}
// logTaskStatusChangeActivity records the task_status_changed activity row
// synchronously, in-process, before the OfficeTaskStatusChanged event is
// published. This used to be an async event-bus subscriber
// (office/service.Service.handleTaskStatusChanged); under a NATS-backed
// bus, that subscriber and the WS broadcast subscriber that triggers the
// frontend's task-detail refetch raced independently, so a browser GET
// could land before the activity row existed and see stale Started/
// Completed sidebar values with nothing left to trigger a corrective
// refetch. Writing the row here, ahead of publishTaskStatusChanged,
// makes the row durable before the event (and therefore the broadcast)
// can ever fire, regardless of bus implementation. Best-effort: mirrors
// logBlockerActivity's activity==nil guard.
func (s *DashboardService) logTaskStatusChangeActivity(ctx context.Context, req TaskStatusUpdateRequest) {
if s.activity == nil {
return
}
wsID, _ := s.repo.GetTaskWorkspaceID(ctx, req.TaskID)
actorType := "system"
actorID := "office-scheduler"
if req.ActorAgentID != "" {
actorType = activityActorTypeAgent
actorID = req.ActorAgentID
}
runID := ""
if s.runResolver != nil {
runID = s.runResolver.ResolveRunForTask(ctx, req.TaskID)
}
s.activity.LogActivityWithRun(ctx, wsID, actorType, actorID,
"task_status_changed", "task", req.TaskID,
fmt.Sprintf(`{"new_status":%q}`, req.NewStatus), runID, "")
}
// applyApprovalGate redirects a "done" transition to in_review when
// approvals are pending. Mutates dbState and apiStatus in place so the
// rest of UpdateTaskStatus persists and emits the redirected status.
@@ -0,0 +1,147 @@
package dashboard_test
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/office/dashboard"
taskmodels "github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// hasStatusChangeActivity reports whether the office_activity_log already
// contains a task_status_changed row for taskID.
func hasStatusChangeActivity(t *testing.T, deps *testDeps, taskID, wsID string) bool {
t.Helper()
entries, err := deps.repo.ListActivityEntriesByTarget(context.Background(), wsID, taskID, 50)
if err != nil {
t.Fatalf("list activity: %v", err)
}
for _, e := range entries {
if e.Action == "task_status_changed" {
return true
}
}
return false
}
// TestUpdateTaskStatus_LogsActivitySynchronously pins that UpdateTaskStatus
// writes the task_status_changed activity row itself, in-process, rather
// than depending on an event-bus subscriber to do it. This DashboardService
// (built by newTestDeps) has no event bus wired at all, so before the fix
// (the write lived solely in office/service.Service.handleTaskStatusChanged,
// an OfficeTaskStatusChanged subscriber) this would never produce a row: the
// event is never published without a bus, and nothing bridges to the
// subscriber outside DashboardService anyway. If this activity write ever
// moves back behind an event-bus round-trip, this test catches it.
func TestUpdateTaskStatus_LogsActivitySynchronously(t *testing.T) {
deps := newTestDeps(t)
insertTestTask(t, deps.db, "sync-task", "ws-sync", "Sync Task", "todo", 0)
if err := deps.svc.UpdateTaskStatus(context.Background(), dashboard.TaskStatusUpdateRequest{
TaskID: "sync-task",
NewStatus: "in_progress",
}); err != nil {
t.Fatalf("update task status: %v", err)
}
if !hasStatusChangeActivity(t, deps, "sync-task", "ws-sync") {
t.Fatal("expected task_status_changed activity row immediately after UpdateTaskStatus returns")
}
}
// LogTaskStateChange is the task-service seam used by workflow moves. It must
// write the same durable projection as the Office status endpoint.
func TestLogTaskStateChange_LogsOfficeTask(t *testing.T) {
deps := newTestDeps(t)
insertTestTask(t, deps.db, "workflow-task", "ws-workflow", "Workflow Task", "in_progress", 0)
if _, err := deps.db.Exec(`INSERT INTO workspaces (id, office_workflow_id) VALUES ('ws-workflow', 'wf-office')`); err != nil {
t.Fatalf("create Office workspace: %v", err)
}
if _, err := deps.db.Exec(`UPDATE tasks SET workflow_id = 'wf-office' WHERE id = 'workflow-task'`); err != nil {
t.Fatalf("attach task to Office workflow: %v", err)
}
deps.svc.LogTaskStateChange(context.Background(), &taskmodels.Task{
ID: "workflow-task", WorkspaceID: "ws-workflow", State: v1.TaskStateCompleted,
}, v1.TaskStateInProgress)
if !hasStatusChangeActivity(t, deps, "workflow-task", "ws-workflow") {
t.Fatal("expected task_status_changed activity row for a workflow state transition")
}
}
// orderingEventBus is a minimal bus.EventBus fake that, on Publish of
// OfficeTaskStatusChanged, snapshots whether the activity row is already
// persisted. Every other EventBus method is a harmless no-op; this fix
// only touches the Publish ordering relative to the activity write.
type orderingEventBus struct {
checkRow func() bool
activityPresentAtPublish bool
published bool
}
func (b *orderingEventBus) Publish(_ context.Context, subject string, _ *bus.Event) error {
if subject == events.OfficeTaskStatusChanged {
b.published = true
b.activityPresentAtPublish = b.checkRow()
}
return nil
}
func (b *orderingEventBus) Subscribe(string, bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (b *orderingEventBus) QueueSubscribe(string, string, bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (b *orderingEventBus) Request(context.Context, string, *bus.Event, time.Duration) (*bus.Event, error) {
return nil, nil
}
func (b *orderingEventBus) Close() {}
func (b *orderingEventBus) IsConnected() bool { return true }
// TestUpdateTaskStatus_LogsActivityBeforePublishingEvent reproduces the P1
// race from PR fixup round 1: under a NATS-backed bus, the activity-log
// write and the WS broadcast that triggers the frontend's task-detail
// refetch were two independent async subscribers to the same
// OfficeTaskStatusChanged event, with no ordering guarantee between them —
// a browser GET could land before the activity row existed and see stale
// Started/Completed values with nothing left to trigger a corrective
// refetch. The fix sequences the activity write before the publish call
// inside UpdateTaskStatus itself, so the row is guaranteed durable before
// the event (and therefore any broadcast subscriber) can ever fire,
// independent of the bus implementation. This test proves that ordering
// directly: by the time Publish(OfficeTaskStatusChanged) is observed, the
// activity row must already be queryable.
func TestUpdateTaskStatus_LogsActivityBeforePublishingEvent(t *testing.T) {
deps := newTestDeps(t)
insertTestTask(t, deps.db, "race-task", "ws-race", "Race Task", "todo", 0)
fake := &orderingEventBus{}
fake.checkRow = func() bool {
return hasStatusChangeActivity(t, deps, "race-task", "ws-race")
}
deps.svc.SetEventBus(fake)
if err := deps.svc.UpdateTaskStatus(context.Background(), dashboard.TaskStatusUpdateRequest{
TaskID: "race-task",
NewStatus: "in_progress",
}); err != nil {
t.Fatalf("update task status: %v", err)
}
if !fake.published {
t.Fatal("expected OfficeTaskStatusChanged to be published")
}
if !fake.activityPresentAtPublish {
t.Fatal("activity row was not yet persisted when the status-changed event was published: " +
"the WS broadcast (and any refetch it triggers) could race ahead of the activity write")
}
}
@@ -0,0 +1,55 @@
package dashboard
import "strings"
func buildStatusTimeline(changes []TimelineEvent) []TimelineEventDTO {
timeline := make([]TimelineEventDTO, len(changes))
for i, ev := range changes {
timeline[i] = TimelineEventDTO{
Type: "status_change",
From: ev.From,
To: ev.To,
At: ev.At,
}
}
return timeline
}
// timelineStatusIsInProgress and timelineStatusIsDone classify a timeline
// event's "to" value into the office in_progress/done status buckets. The
// activity log can contain API statuses, persisted states, or raw step names.
func timelineStatusIsInProgress(raw string) bool {
switch strings.ToLower(strings.TrimSpace(raw)) {
case statusInProgressLowercase, "in progress":
return true
default:
return false
}
}
func timelineStatusIsDone(raw string) bool {
switch strings.ToLower(strings.TrimSpace(raw)) {
case statusDoneLowercase, "completed":
return true
default:
return false
}
}
// deriveTaskTimestamps derives Started and Completed timestamps from the
// status-change timeline. Callers clear Completed when the task is reopened.
func deriveTaskTimestamps(changes []TimelineEvent) (startedAt, completedAt string) {
for _, ev := range changes {
switch {
case timelineStatusIsInProgress(ev.To):
if startedAt == "" || ev.At < startedAt {
startedAt = ev.At
}
case timelineStatusIsDone(ev.To):
if ev.At > completedAt {
completedAt = ev.At
}
}
}
return startedAt, completedAt
}
@@ -0,0 +1,93 @@
package dashboard
import "testing"
// TestDeriveTaskTimestamps locks down the Started/Completed sidebar
// derivation described in docs/specs/office/tasks.md: startedAt is the
// earliest transition into in_progress, completedAt is the latest
// transition into done. A regression here silently reverts the Office
// task detail sidebar to permanently blank Started/Completed rows.
func TestDeriveTaskTimestamps(t *testing.T) {
cases := []struct {
name string
changes []TimelineEvent
wantStarted string
wantCompleted string
}{
{
name: "no status changes",
changes: nil,
wantStarted: "",
wantCompleted: "",
},
{
name: "todo to in_progress only",
changes: []TimelineEvent{
{From: "todo", To: "in_progress", At: "2026-08-01T10:00:00Z"},
},
wantStarted: "2026-08-01T10:00:00Z",
wantCompleted: "",
},
{
name: "full todo -> in_progress -> done",
changes: []TimelineEvent{
// ListActivityEntriesByTarget returns rows DESC by created_at.
{From: "in_progress", To: "done", At: "2026-08-01T12:00:00Z"},
{From: "todo", To: "in_progress", At: "2026-08-01T10:00:00Z"},
},
wantStarted: "2026-08-01T10:00:00Z",
wantCompleted: "2026-08-01T12:00:00Z",
},
{
name: "reopened: done -> in_progress -> done, expect latest done",
changes: []TimelineEvent{
{From: "in_progress", To: "done", At: "2026-08-03T09:00:00Z"},
{From: "done", To: "in_progress", At: "2026-08-02T08:00:00Z"},
{From: "in_progress", To: "done", At: "2026-08-01T12:00:00Z"},
{From: "todo", To: "in_progress", At: "2026-08-01T10:00:00Z"},
},
wantStarted: "2026-08-01T10:00:00Z",
wantCompleted: "2026-08-03T09:00:00Z",
},
{
name: "out-of-order input still resolves min/max",
changes: []TimelineEvent{
{From: "todo", To: "in_progress", At: "2026-08-05T10:00:00Z"},
{From: "in_progress", To: "done", At: "2026-08-01T12:00:00Z"},
{From: "todo", To: "in_progress", At: "2026-08-01T09:00:00Z"},
},
wantStarted: "2026-08-01T09:00:00Z",
wantCompleted: "2026-08-01T12:00:00Z",
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
gotStarted, gotCompleted := deriveTaskTimestamps(c.changes)
if gotStarted != c.wantStarted {
t.Errorf("startedAt = %q, want %q", gotStarted, c.wantStarted)
}
if gotCompleted != c.wantCompleted {
t.Errorf("completedAt = %q, want %q", gotCompleted, c.wantCompleted)
}
})
}
}
// TestDeriveTaskTimestampsStepNameVariants covers the workflow-engine
// TaskMoved producer, whose activity entries carry raw step names ("In
// Progress", "Done") rather than the already-normalized lowercase status
// strings the direct Office status-update path writes. Both producers
// write the same "task_status_changed" activity action.
func TestDeriveTaskTimestampsStepNameVariants(t *testing.T) {
changes := []TimelineEvent{
{From: "In Progress", To: "Done", At: "2026-08-01T12:00:00Z"},
{From: "Todo", To: "In Progress", At: "2026-08-01T10:00:00Z"},
}
gotStarted, gotCompleted := deriveTaskTimestamps(changes)
if gotStarted != "2026-08-01T10:00:00Z" {
t.Errorf("startedAt = %q, want 2026-08-01T10:00:00Z", gotStarted)
}
if gotCompleted != "2026-08-01T12:00:00Z" {
t.Errorf("completedAt = %q, want 2026-08-01T12:00:00Z", gotCompleted)
}
}
@@ -90,15 +90,6 @@ type ApprovalResolvedData struct {
Type string `json:"type"`
}
// TaskStatusChangedData represents the payload of an office.task.status_changed event.
type TaskStatusChangedData struct {
TaskID string `json:"task_id"`
WorkspaceID string `json:"workspace_id"`
NewStatus string `json:"new_status"`
Comment string `json:"comment"`
ActorAgentID string `json:"actor_agent_id"`
}
// AgentLifecycleData is the subset of agent lifecycle event data needed by office.
//
// AgentID is populated by the lifecycle manager for taskless runs
@@ -219,7 +210,6 @@ func (s *Service) RegisterEventSubscribers(eb bus.EventBus) error {
{events.TaskMoved, s.handleTaskMoved},
{events.OfficeApprovalResolved, s.handleApprovalResolved},
{events.OfficeCommentCreated, s.handleCommentCreated},
{events.OfficeTaskStatusChanged, s.handleTaskStatusChanged},
{events.AgentCompleted, maybeAsync(s.handleAgentCompleted)},
// AgentStopped fires when StopAgent is called (e.g. office
// fire-and-forget turn-complete teardown). Same handler — both
@@ -844,12 +834,13 @@ func (s *Service) queueTaskAssignedRun(
return s.QueueRun(ctx, agentProfileID, RunReasonTaskAssigned, payload, key)
}
// handleTaskMoved logs the step change to the activity log and queues
// handleTaskMoved keeps the legacy named-step activity fallback and queues
// downstream blocker / children-completed runs when a task lands in a
// terminal step. Stage progression itself is owned by the workflow
// engine (the orchestrator subscribes to TaskMoved and fires
// on_exit / on_enter); this handler covers the side-effects the engine
// path doesn't yet emit.
// terminal step. Canonical task-state activity is written by the task service
// before task.state_changed is published, so the workflow move path is durable
// before any WebSocket refetch can run. Stage progression itself is owned by
// the workflow engine (the orchestrator subscribes to TaskMoved and fires
// on_exit / on_enter).
func (s *Service) handleTaskMoved(ctx context.Context, event *bus.Event) error {
data, err := decodeEventData[TaskMovedData](event)
if err != nil || data.AssigneeAgentProfileID == "" {
@@ -996,33 +987,6 @@ func (s *Service) queueChildrenCompletedRun(ctx context.Context, parentID string
engine.OnChildrenCompletedPayload{ChildSummaries: summaries}, key)
}
// handleTaskStatusChanged logs status-change activity. Stage progression
// (Work → Review → Approval → Done) is owned by the workflow engine via
// the on_exit / on_enter triggers fired by the orchestrator on
// TaskMoved; the legacy ExecutionPolicy transition path was removed in
// Phase 4 of task-model-unification.
func (s *Service) handleTaskStatusChanged(ctx context.Context, event *bus.Event) error {
data, err := decodeEventData[TaskStatusChangedData](event)
if err != nil || data.TaskID == "" || data.NewStatus == "" {
return nil
}
fields, _ := s.repo.GetTaskExecutionFields(ctx, data.TaskID)
if fields != nil && fields.WorkspaceID != "" {
actorType := "system"
actorID := "office-scheduler"
if data.ActorAgentID != "" {
actorType = participantTypeAgent
actorID = data.ActorAgentID
}
runID := s.ResolveRunForTask(ctx, data.TaskID)
s.LogActivityWithRun(ctx, fields.WorkspaceID, actorType, actorID,
"task_status_changed", "task", data.TaskID,
fmt.Sprintf(`{"new_status":%q}`, data.NewStatus), runID, "")
}
return nil
}
// handleCommentCreated loads the comment and relays it to external channels.
func (s *Service) handleCommentCreated(ctx context.Context, event *bus.Event) error {
data, err := decodeEventData[CommentPostedData](event)
@@ -783,38 +783,6 @@ func TestMovedToDone_WithoutExecutionPolicy_NormalCompletion(t *testing.T) {
}
}
func TestHandleTaskStatusChanged_LogsActivity(t *testing.T) {
svc, eb := newTestServiceWithBus(t)
ctx := context.Background()
createTestAgent(t, svc, "ws-1", "worker-status")
taskID := createOfficeTask(t, svc, "ws-1", "worker-status")
statusEvt := bus.NewEvent(events.OfficeTaskStatusChanged, "test", map[string]interface{}{
"task_id": taskID,
"new_status": "in_progress",
"workspace_id": "ws-1",
})
if err := eb.Publish(ctx, events.OfficeTaskStatusChanged, statusEvt); err != nil {
t.Fatalf("publish office.task.status_changed: %v", err)
}
activity, err := svc.ListActivity(ctx, "ws-1", 50)
if err != nil {
t.Fatalf("list activity: %v", err)
}
found := false
for _, a := range activity {
if a.Action == "task_status_changed" && a.TargetID == taskID {
found = true
break
}
}
if !found {
t.Errorf("expected activity entry with action 'task_status_changed' for task %s, got %+v", taskID, activity)
}
}
func TestAutoPostAgentComment_CreatesSessionComment(t *testing.T) {
svc, eb := newTestServiceWithBus(t)
ctx := context.Background()
@@ -344,6 +344,7 @@ type Service struct {
// other's insert and commit a cycle between them.
dependencyEdgeMu sync.Mutex
comments CommentRepository
taskStateActivity TaskStateActivityLogger
secretStore secrets.SecretStore
workspaceSecretDeleter WorkspaceSecretDeleter
baseBranchPusher AgentBaseBranchPusher
@@ -440,6 +440,12 @@ func (s *Service) publishTaskEventNow(ctx context.Context, eventType string, tas
for k, v := range extra {
data[k] = v
}
if eventType == events.TaskStateChanged && oldState != nil && s.taskStateActivity != nil {
// Write the Office read-model row before publishing the event. The
// WebSocket broadcaster can then trigger a detail GET without racing
// the activity projection that supplies Started and Completed.
s.taskStateActivity.LogTaskStateChange(ctx, task, *oldState)
}
event := bus.NewEvent(eventType, "task-service", data)
err := s.eventBus.Publish(ctx, eventType, event)
@@ -8,6 +8,7 @@ import (
orchmodels "github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// BlockerRepository provides access to task blocker persistence.
@@ -36,6 +37,13 @@ type CommentRepository interface {
ListTaskComments(ctx context.Context, taskID string) ([]*orchmodels.TaskComment, error)
}
// TaskStateActivityLogger records a durable status transition before the task
// state event is published. Optional callers can use this seam to keep
// read-model activity and WebSocket notifications ordered.
type TaskStateActivityLogger interface {
LogTaskStateChange(ctx context.Context, task *models.Task, oldState v1.TaskState)
}
// SetBlockerRepository wires the blocker repository for office integration.
func (s *Service) SetBlockerRepository(repo BlockerRepository) {
s.blockers = repo
@@ -46,6 +54,12 @@ func (s *Service) SetCommentRepository(repo CommentRepository) {
s.comments = repo
}
// SetTaskStateActivityLogger wires the optional durable activity writer used
// before task.state_changed notifications are published.
func (s *Service) SetTaskStateActivityLogger(logger TaskStateActivityLogger) {
s.taskStateActivity = logger
}
// GetLastAgentMessage returns the content of the most recent agent message
// in a session. Used by the office comment bridge to auto-post agent responses.
func (s *Service) GetLastAgentMessage(ctx context.Context, sessionID string) (string, error) {
@@ -121,6 +121,47 @@ func TestStandaloneTaskEventPublishers(t *testing.T) {
}
}
type recordingTaskStateActivityLogger struct {
called bool
taskID string
old v1.TaskState
new v1.TaskState
publishedBefore int
eventBus *MockEventBus
}
func (l *recordingTaskStateActivityLogger) LogTaskStateChange(
_ context.Context, task *models.Task, oldState v1.TaskState,
) {
l.called = true
l.taskID = task.ID
l.old = oldState
l.new = task.State
if l.eventBus != nil {
l.publishedBefore = len(l.eventBus.GetPublishedEvents())
}
}
func TestTaskStateActivityIsLoggedBeforeStateEvent(t *testing.T) {
svc, eventBus, _ := createTestService(t)
activity := &recordingTaskStateActivityLogger{eventBus: eventBus}
svc.SetTaskStateActivityLogger(activity)
svc.PublishTaskStateChanged(context.Background(), &models.Task{
ID: "task-state-activity", WorkspaceID: "ws-1", State: v1.TaskStateCompleted,
}, v1.TaskStateInProgress)
if !activity.called || activity.taskID != "task-state-activity" {
t.Fatalf("activity logger call = %+v, want task-state-activity", activity)
}
if activity.old != v1.TaskStateInProgress || activity.new != v1.TaskStateCompleted {
t.Fatalf("activity transition = %q -> %q", activity.old, activity.new)
}
if activity.publishedBefore != 0 {
t.Fatalf("activity logger observed %d published events, want 0", activity.publishedBefore)
}
}
func TestPublishAfterTaskEventsRunsCallback(t *testing.T) {
svc, _, _ := createTestService(t)
ctx := context.Background()
@@ -0,0 +1,61 @@
import { describe, expect, it } from "vitest";
import type { OfficeTask } from "@/lib/state/slices/office/types";
import { mapOfficeTaskToTask } from "./map-office-task";
function baseOfficeTask(overrides: Partial<OfficeTask> = {}): OfficeTask {
return {
id: "task-1",
workspaceId: "ws-1",
identifier: "OFF-1",
title: "Test task",
status: "in_progress",
priority: "medium",
createdAt: "2026-08-01T09:00:00Z",
updatedAt: "2026-08-01T09:00:00Z",
...overrides,
};
}
describe("mapOfficeTaskToTask", () => {
it("maps startedAt and completedAt through when the backend sends them", () => {
const raw = baseOfficeTask({
startedAt: "2026-08-01T10:00:00Z",
completedAt: "2026-08-01T12:00:00Z",
});
const task = mapOfficeTaskToTask(raw);
expect(task.startedAt).toBe("2026-08-01T10:00:00Z");
expect(task.completedAt).toBe("2026-08-01T12:00:00Z");
});
it("leaves startedAt and completedAt undefined when the backend omits them", () => {
const raw = baseOfficeTask();
const task = mapOfficeTaskToTask(raw);
expect(task.startedAt).toBeUndefined();
expect(task.completedAt).toBeUndefined();
});
it("always maps createdBy to an empty string (no source of truth yet)", () => {
const task = mapOfficeTaskToTask(baseOfficeTask());
expect(task.createdBy).toBe("");
});
// The detail page seeds its Task from the store-held OfficeTask before the
// API GET resolves. Preserve rawStatus so execution indicators keep their
// live/ready distinction during that initial render.
it("carries rawStatus through when the store holds one", () => {
const mapped = mapOfficeTaskToTask(baseOfficeTask({ status: "todo", rawStatus: "SCHEDULING" }));
expect(mapped.rawStatus).toBe("SCHEDULING");
});
it("falls back to the canonical status when rawStatus is absent", () => {
const mapped = mapOfficeTaskToTask(baseOfficeTask({ status: "in_progress" }));
expect(mapped.rawStatus).toBe("in_progress");
});
});
@@ -0,0 +1,65 @@
import type { TaskDecisionDTO } from "@/lib/api/domains/office-api";
import type { OfficeTask } from "@/lib/state/slices/office/types";
import type { Task, TaskDecision } from "./types";
export function mapDecisionDTO(d: TaskDecisionDTO): TaskDecision {
return {
id: d.id,
taskId: d.task_id,
deciderType: d.decider_type,
deciderId: d.decider_id,
deciderName: d.decider_name ?? "",
role: d.role,
decision: d.decision,
comment: d.comment ?? "",
createdAt: d.created_at,
};
}
export function mapOfficeTaskToTask(raw: OfficeTask): Task {
// The server DTO includes reviewers/approvers/decisions even though the
// strongly-typed OfficeTask only declares the cross-cutting fields. We
// read those extra props off the raw object.
const extra = raw as OfficeTask & {
reviewers?: string[];
approvers?: string[];
decisions?: TaskDecisionDTO[];
blockedBy?: string[];
};
return {
id: raw.id,
workspaceId: raw.workspaceId,
identifier: raw.identifier,
title: raw.title,
description: raw.description,
status: raw.status as Task["status"],
rawStatus: raw.rawStatus ?? raw.status,
priority: (raw.priority || "medium") as Task["priority"],
labels: (raw.labels ?? []).map((l) =>
typeof l === "string" ? { name: l, color: "#6b7280" } : l,
),
assigneeAgentProfileId: raw.assigneeAgentProfileId,
parentId: raw.parentId,
projectId: raw.projectId,
blockedBy: extra.blockedBy ?? [],
blocking: [],
children: (raw.children ?? []).map((child) => ({
id: child.id,
identifier: child.identifier,
title: child.title,
status: child.status as Task["status"],
blockedBy: child.blockedBy ?? [],
createdAt: child.createdAt,
})),
reviewers: extra.reviewers ?? [],
approvers: extra.approvers ?? [],
decisions: (extra.decisions ?? []).map(mapDecisionDTO),
createdBy: "",
startedAt: raw.startedAt,
completedAt: raw.completedAt,
createdAt: raw.createdAt,
updatedAt: raw.updatedAt,
executionPolicy: raw.executionPolicy,
executionState: raw.executionState,
};
}
@@ -1,33 +0,0 @@
import { describe, expect, it } from "vitest";
import { mapOfficeTaskToTask } from "./page";
import type { OfficeTask } from "@/lib/state/slices/office/types";
const baseRaw: OfficeTask = {
id: "task-1",
workspaceId: "workspace-1",
identifier: "E2E-1",
title: "Mapper regression task",
status: "todo",
priority: "medium",
createdAt: "2026-05-01T10:00:00Z",
updatedAt: "2026-05-01T10:00:00Z",
};
describe("mapOfficeTaskToTask", () => {
// Regression: the detail page seeds its Task from the store-held OfficeTask
// before the API GET resolves (page.tsx's useIssueData). Dropping rawStatus
// here silently breaks ExecutionIndicator's Live/Ready sub-state distinction
// for however long the seeded task is shown — see task-advanced-mode.test.tsx
// and OfficeSimplePane.test.tsx for the consumer-level regression.
it("carries rawStatus through when the store holds one", () => {
const mapped = mapOfficeTaskToTask({ ...baseRaw, status: "todo", rawStatus: "SCHEDULING" });
expect(mapped.rawStatus).toBe("SCHEDULING");
});
it("falls back to the canonical status when rawStatus is absent", () => {
const mapped = mapOfficeTaskToTask({ ...baseRaw, status: "in_progress" });
expect(mapped.rawStatus).toBe("in_progress");
});
});
+15 -71
View File
@@ -4,13 +4,13 @@ import { use, useState, useEffect, useRef, useCallback, useMemo, Suspense } from
import { useRouter, useSearchParams } from "@/lib/routing/client-router";
import { useAppStore } from "@/components/state-provider";
import { useOfficeRefetch } from "@/hooks/use-office-refetch";
import { useLatestOnly } from "@/hooks/use-latest-only";
import { TaskOptimisticContextProvider } from "@/hooks/use-optimistic-task-mutation";
import {
getTask,
listActivityForTarget,
listComments,
type TaskCommentResponse,
type TaskDecisionDTO,
} from "@/lib/api/domains/office-api";
import { listTaskSessions } from "@/lib/api/domains/session-api";
import {
@@ -21,19 +21,13 @@ import { OfficeSimplePane } from "@/components/task/simple/OfficeSimplePane";
import { TaskAdvancedMode } from "./task-advanced-mode";
import { IssueDetailSkeleton } from "./task-detail-skeleton";
import { TaskBody, resolveTaskBodyMode, type TaskBodyMode } from "@/components/task/TaskBody";
import type {
Task,
TaskComment,
TaskActivityEntry,
TaskDecision,
TaskSession,
TimelineEvent,
} from "./types";
import type { ActivityEntry, OfficeTask } from "@/lib/state/slices/office/types";
import type { Task, TaskComment, TaskActivityEntry, TaskSession, TimelineEvent } from "./types";
import type { ActivityEntry } from "@/lib/state/slices/office/types";
import type { TaskSession as ApiTaskSession } from "@/lib/types/http";
import { useSessionLiveSyncSubscriptions } from "./use-session-live-sync";
import { useTranslation } from "react-i18next";
import { t } from "@/lib/i18n";
import { mapOfficeTaskToTask } from "./map-office-task";
type IssueDetailPageProps = {
params: Promise<{ id: string }>;
@@ -44,66 +38,6 @@ type IssueDetailPageProps = {
// cleared and must not be resurrected).
const SESSION_METADATA_ABSENT = "\u0000absent\u0000";
function mapDecisionDTO(d: TaskDecisionDTO): TaskDecision {
return {
id: d.id,
taskId: d.task_id,
deciderType: d.decider_type,
deciderId: d.decider_id,
deciderName: d.decider_name ?? "",
role: d.role,
decision: d.decision,
comment: d.comment ?? "",
createdAt: d.created_at,
};
}
export function mapOfficeTaskToTask(raw: OfficeTask): Task {
// The server DTO includes reviewers/approvers/decisions even though the
// strongly-typed OfficeTask only declares the cross-cutting fields. We
// read those extra props off the raw object.
const extra = raw as OfficeTask & {
reviewers?: string[];
approvers?: string[];
decisions?: TaskDecisionDTO[];
blockedBy?: string[];
};
return {
id: raw.id,
workspaceId: raw.workspaceId,
identifier: raw.identifier,
title: raw.title,
description: raw.description,
status: raw.status as Task["status"],
rawStatus: raw.rawStatus ?? raw.status,
priority: (raw.priority || "medium") as Task["priority"],
labels: (raw.labels ?? []).map((l) =>
typeof l === "string" ? { name: l, color: "#6b7280" } : l,
),
assigneeAgentProfileId: raw.assigneeAgentProfileId,
parentId: raw.parentId,
projectId: raw.projectId,
blockedBy: extra.blockedBy ?? [],
blocking: [],
children: (raw.children ?? []).map((child) => ({
id: child.id,
identifier: child.identifier,
title: child.title,
status: child.status as Task["status"],
blockedBy: child.blockedBy ?? [],
createdAt: child.createdAt,
})),
reviewers: extra.reviewers ?? [],
approvers: extra.approvers ?? [],
decisions: (extra.decisions ?? []).map(mapDecisionDTO),
createdBy: "",
createdAt: raw.createdAt,
updatedAt: raw.updatedAt,
executionPolicy: raw.executionPolicy,
executionState: raw.executionState,
};
}
function mapCommentResponse(c: TaskCommentResponse): TaskComment {
return {
id: c.id,
@@ -316,9 +250,19 @@ function useTaskOptimisticHelpers(
// (priority / project / parent / blockers / participants / assignee).
// The optimistic patch we applied locally gets reconciled with server
// state.
//
// A WS-driven trigger can fire again before a prior GET resolves (two
// status changes in quick succession, ordinary network jitter). Without
// a guard, a slower earlier response can land after a faster later one
// and overwrite fresher task state with stale data. useLatestOnly
// discards a response once a newer refetchTask call has started,
// regardless of arrival order.
const { begin, isCurrent } = useLatestOnly();
const refetchTask = useCallback(async () => {
const token = begin();
try {
const res = await getTask(id);
if (!isCurrent(token)) return;
if (res.task) {
setTask(mapOfficeTaskToTask(res.task));
if (res.timeline) setTimeline(res.timeline);
@@ -326,7 +270,7 @@ function useTaskOptimisticHelpers(
} catch {
/* swallow — next user action will retry */
}
}, [id, setTask, setTimeline]);
}, [id, setTask, setTimeline, begin, isCurrent]);
useOfficeRefetch(`task:${id}`, () => {
void refetchTask();
});
@@ -185,7 +185,7 @@ export function ApprovalActionBar({ task }: ApprovalActionBarProps) {
// optimistic id with the real decision row id is performed implicitly
// when `task:<id>` refetches via the WS handler. We keep the helper
// available so future flows (e.g. a per-decision detail view) can map
// the wire DTO consistently with mapDecisionDTO in page.tsx.
// the wire DTO consistently with mapDecisionDTO in map-office-task.ts.
export { dtoToDecision };
type ApprovalCommentFormProps = {
@@ -9,6 +9,8 @@ type PropertyRowProps = {
valueClassName?: string;
/** When true, align children to the top instead of centred. */
alignStart?: boolean;
/** Test id placed on the value side, for E2E locators. */
testId?: string;
};
/**
@@ -22,6 +24,7 @@ export function PropertyRow({
className,
valueClassName,
alignStart = false,
testId,
}: PropertyRowProps) {
return (
<div
@@ -32,7 +35,9 @@ export function PropertyRow({
)}
>
<span className="text-sm text-muted-foreground w-24 shrink-0 pt-0.5">{label}</span>
<div className={cn("text-sm text-right flex-1 min-w-0", valueClassName)}>{children}</div>
<div className={cn("text-sm text-right flex-1 min-w-0", valueClassName)} data-testid={testId}>
{children}
</div>
</div>
);
}
@@ -116,11 +116,18 @@ function TimelineSection({ task }: { task: Task }) {
const { t } = useTranslation();
const dateOrDash = (d?: string | null) =>
d ? formatDate(d) : <span className="text-muted-foreground">--</span>;
const orDash = (v?: string | null) => (v ? v : <span className="text-muted-foreground">--</span>);
return (
<>
<PropertyRow label={t("task:createdBy")}>{task.createdBy}</PropertyRow>
<PropertyRow label={t("task:started")}>{dateOrDash(task.startedAt)}</PropertyRow>
<PropertyRow label={t("task:completed")}>{dateOrDash(task.completedAt)}</PropertyRow>
<PropertyRow label={t("task:createdBy")} testId="created-by-row">
{orDash(task.createdBy)}
</PropertyRow>
<PropertyRow label={t("task:started")} testId="started-row">
{dateOrDash(task.startedAt)}
</PropertyRow>
<PropertyRow label={t("task:completed")} testId="completed-row">
{dateOrDash(task.completedAt)}
</PropertyRow>
<PropertyRow label={t("task:created")}>{formatDate(task.createdAt)}</PropertyRow>
<PropertyRow label={t("task:updated")}>{formatRelativeTime(task.updatedAt)}</PropertyRow>
</>
@@ -1,5 +1,6 @@
import { test, expect } from "../../fixtures/office-fixture";
import type { Page } from "@playwright/test";
import { waitForHttp } from "../../helpers/causal-waits";
/**
* E2E coverage for office task property pickers (status, priority,
@@ -360,4 +361,53 @@ test.describe("property pickers", () => {
expect(after).toBe(before);
}).toPass({ timeout: 5_000 });
});
test("started and completed rows show timestamps after a todo -> in_progress -> done transition", async ({
testPage,
apiClient,
officeSeed,
}) => {
const task = await apiClient.createTask(officeSeed.workspaceId, "Picker Timeline Task", {
workflow_id: officeSeed.workflowId,
});
// startedAt/completedAt are derived server-side from the status-change
// timeline (see deriveTaskTimestamps in the backend) and delivered to
// the open page by the "office.task.status_changed" WS handler's
// per-task refetch (GET /api/v1/office/tasks/:id) — assert on that
// live-page refetch rather than a reload, so a regression of the
// refetch itself fails this test.
const taskDetailPath = new RegExp(`^/api/v1/office/tasks/${task.id}$`);
// gotoTaskPage only waits for the title heading, which can render before
// the sidebar's own render pass picks up the same task fetch; wait for
// the initial GET explicitly instead of racing the property rows against
// an unbudgeted paint (see e2e/helpers/causal-waits.ts).
const initialLoad = waitForHttp(testPage, "GET", taskDetailPath);
await gotoTaskPage(testPage, task.id, "Picker Timeline Task");
await initialLoad;
const startedRow = testPage.getByTestId("started-row");
const completedRow = testPage.getByTestId("completed-row");
await expect(startedRow).toHaveText("--");
await expect(completedRow).toHaveText("--");
const startedRefetch = waitForHttp(testPage, "GET", taskDetailPath);
await testPage.getByTestId("status-picker-trigger").click();
await testPage.getByTestId("status-picker-option-in_progress").click();
await expect(testPage.getByTestId("status-picker-trigger")).toContainText(/In Progress/i, {
timeout: 15_000,
});
await startedRefetch;
await expect(startedRow).not.toHaveText("--", { timeout: 5_000 });
const completedRefetch = waitForHttp(testPage, "GET", taskDetailPath);
await testPage.getByTestId("status-picker-trigger").click();
await testPage.getByTestId("status-picker-option-done").click();
await expect(testPage.getByTestId("status-picker-trigger")).toContainText(/Done/i, {
timeout: 15_000,
});
await completedRefetch;
await expect(completedRow).not.toHaveText("--", { timeout: 5_000 });
});
});
+46
View File
@@ -0,0 +1,46 @@
import { describe, expect, it } from "vitest";
import { renderHook, act } from "@testing-library/react";
import { useLatestOnly } from "./use-latest-only";
// Regression for the Office task-detail refetch race: a WS trigger can
// fire again before a prior GET resolves, and a slower earlier response
// can arrive after a faster later one. useLatestOnly must let the caller
// discard the stale response regardless of arrival order.
describe("useLatestOnly", () => {
it("marks only the most recently begun token as current", () => {
const { result } = renderHook(() => useLatestOnly());
let tokenA = -1;
let tokenB = -1;
act(() => {
tokenA = result.current.begin();
tokenB = result.current.begin();
});
expect(result.current.isCurrent(tokenA)).toBe(false);
expect(result.current.isCurrent(tokenB)).toBe(true);
});
it("discards an earlier call's result when it resolves after a later call started", async () => {
const { result } = renderHook(() => useLatestOnly());
const applied: string[] = [];
// Simulates the exact race: call A starts, call B starts before A
// resolves, then A's (stale) response arrives followed by B's.
const tokenA = result.current.begin();
const tokenB = result.current.begin();
if (result.current.isCurrent(tokenA)) applied.push("A");
if (result.current.isCurrent(tokenB)) applied.push("B");
expect(applied).toEqual(["B"]);
});
it("accepts a call's result when no newer call has started since", () => {
const { result } = renderHook(() => useLatestOnly());
const token = result.current.begin();
expect(result.current.isCurrent(token)).toBe(true);
});
});
+16
View File
@@ -0,0 +1,16 @@
import { useCallback, useRef } from "react";
/**
* Ordering guard for async work that can overlap (a WS trigger firing
* again before a prior GET resolves, ordinary network jitter). `begin()`
* stamps a new generation before starting the work; `isCurrent(token)`
* reports whether no newer call has started since — even if this call's
* response arrives after a more recent call's response, so callers can
* discard it instead of overwriting fresher state with stale data.
*/
export function useLatestOnly() {
const generationRef = useRef(0);
const begin = useCallback(() => ++generationRef.current, []);
const isCurrent = useCallback((token: number) => token === generationRef.current, []);
return { begin, isCurrent };
}
+144
View File
@@ -0,0 +1,144 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { cleanup, renderHook } from "@testing-library/react";
let storeState: { office: { refetchTriggers: Record<string, number> } } = {
office: { refetchTriggers: {} },
};
vi.mock("@/components/state-provider", () => ({
useAppStore: (selector: (state: typeof storeState) => unknown) => selector(storeState),
}));
import { useOfficeRefetch } from "./use-office-refetch";
function bump(type: string) {
const triggers = storeState.office.refetchTriggers;
storeState = {
office: { refetchTriggers: { ...triggers, [type]: (triggers[type] ?? 0) + 1 } },
};
}
describe("useOfficeRefetch", () => {
beforeEach(() => {
storeState = { office: { refetchTriggers: {} } };
});
afterEach(() => {
cleanup();
});
it("does not fire on mount when no trigger has fired for this type", () => {
const onRefetch = vi.fn();
renderHook(() => useOfficeRefetch("dashboard", onRefetch));
expect(onRefetch).not.toHaveBeenCalled();
});
it("fires when its exact type is bumped", () => {
const onRefetch = vi.fn();
const { rerender } = renderHook(() => useOfficeRefetch("dashboard", onRefetch));
bump("dashboard");
rerender();
expect(onRefetch).toHaveBeenCalledTimes(1);
});
it("fires on a prefix match (comments: matches comments:task-123)", () => {
const onRefetch = vi.fn();
const { rerender } = renderHook(() => useOfficeRefetch("comments", onRefetch));
bump("comments:task-123");
rerender();
expect(onRefetch).toHaveBeenCalledTimes(1);
});
it("does not fire for an unrelated type", () => {
const onRefetch = vi.fn();
const { rerender } = renderHook(() => useOfficeRefetch("dashboard", onRefetch));
bump("tasks");
rerender();
expect(onRefetch).not.toHaveBeenCalled();
});
it("does not re-fire on a render that doesn't bump its type", () => {
const onRefetch = vi.fn();
const { rerender } = renderHook(() => useOfficeRefetch("dashboard", onRefetch));
bump("dashboard");
rerender();
rerender();
rerender();
expect(onRefetch).toHaveBeenCalledTimes(1);
});
// Regression test: a single WS handler often bumps several distinct trigger
// types in one synchronous call (e.g. `task:${id}` then `dashboard`), which
// React coalesces into a single render. A shared "last trigger" store field
// would overwrite itself on every bump, so only the final type's subscriber
// would ever observe a change — every other type silently never refetches.
// Per-type counters must let every matching subscriber see its own bump
// even when several types were bumped before React ever renders.
it("lets every subscriber observe its own bump when multiple types are bumped in the same tick", () => {
const onTaskRefetch = vi.fn();
const onDashboardRefetch = vi.fn();
const { rerender: rerenderTask } = renderHook(() =>
useOfficeRefetch("task:t-1", onTaskRefetch),
);
const { rerender: rerenderDashboard } = renderHook(() =>
useOfficeRefetch("dashboard", onDashboardRefetch),
);
// Simulate a WS handler firing both triggers before either hook re-renders.
bump("task:t-1");
bump("dashboard");
rerenderTask();
rerenderDashboard();
expect(onTaskRefetch).toHaveBeenCalledTimes(1);
expect(onDashboardRefetch).toHaveBeenCalledTimes(1);
});
// Regression test: a prefix subscriber watches several sub-keys under one
// triggerType (e.g. "comments" matches "comments:task-A" and
// "comments:task-B"). A single scalar "last seen" value would let a bump
// on one sub-key mask a later bump on a sibling sub-key whenever the
// sibling's counter is lower than the max already observed. Per-key
// tracking must let each sub-key's bump fire independently.
it("still fires for a sub-key bump after a higher-numbered sibling sub-key already fired", () => {
const onRefetch = vi.fn();
const { rerender } = renderHook(() => useOfficeRefetch("comments", onRefetch));
// task-B races ahead first, bumped three times.
bump("comments:task-B");
bump("comments:task-B");
bump("comments:task-B");
rerender();
expect(onRefetch).toHaveBeenCalledTimes(1);
// task-A's first bump (counter 1) is lower than task-B's counter (3), so
// a scalar "max seen" comparison would swallow it.
bump("comments:task-A");
rerender();
expect(onRefetch).toHaveBeenCalledTimes(2);
});
it("uses the latest callback without re-firing when only the callback changes", () => {
const first = vi.fn();
const second = vi.fn();
const { rerender } = renderHook(({ cb }) => useOfficeRefetch("dashboard", cb), {
initialProps: { cb: first },
});
rerender({ cb: second });
bump("dashboard");
rerender({ cb: second });
expect(first).not.toHaveBeenCalled();
expect(second).toHaveBeenCalledTimes(1);
});
});
+29 -7
View File
@@ -1,26 +1,48 @@
import { useEffect, useLayoutEffect, useRef } from "react";
import { useAppStore } from "@/components/state-provider";
// Any real counter the store can hold is >= 1, so 0 reliably "looks bumped"
// on the very first observed value and "looks unbumped" otherwise — see
// use-session-commits.ts for the same sentinel pattern.
const REFETCH_TRIGGER_INIT = 0;
/**
* Calls `onRefetch` when the office refetch trigger matches `triggerType`.
* Calls `onRefetch` when a matching office refetch trigger fires.
* Supports exact match ("dashboard") or prefix match ("comments:" matches "comments:task-123").
*
* Triggers are per-type counters (`office.refetchTriggers`), not a single
* "last trigger" value: a WS handler often bumps several distinct types in
* one synchronous call (e.g. `task:${id}` then `dashboard`), and React
* coalesces those into one render, so a shared last-value field would only
* ever be observable by the final type's subscribers.
*
* @param triggerType - The trigger type to watch for (e.g. "dashboard", "tasks", "comments")
* @param onRefetch - Callback invoked when a matching trigger fires
*/
export function useOfficeRefetch(triggerType: string, onRefetch: () => void) {
const trigger = useAppStore((s) => s.office.refetchTrigger);
const triggers = useAppStore((s) => s.office.refetchTriggers);
const callbackRef = useRef(onRefetch);
// Update ref in a layout effect to avoid mutating during render
useLayoutEffect(() => {
callbackRef.current = onRefetch;
});
// Tracks the highest counter we've already acted on, per matching key, so
// a bump triggers exactly one refetch rather than re-firing on every
// render. A single scalar here would let a bump on one sub-key (e.g.
// "comments:task-B") mask a later bump on a sibling sub-key ("comments:
// task-A") whenever the sibling's counter happened to be lower.
const lastSeenRef = useRef<Record<string, number>>({});
useEffect(() => {
if (!trigger) return;
const matches = trigger.type === triggerType || trigger.type.startsWith(triggerType + ":");
if (matches) {
callbackRef.current();
let fire = false;
for (const [type, seq] of Object.entries(triggers)) {
if (type !== triggerType && !type.startsWith(triggerType + ":")) continue;
if (seq > (lastSeenRef.current[type] ?? REFETCH_TRIGGER_INIT)) {
lastSeenRef.current[type] = seq;
fire = true;
}
}
}, [trigger, triggerType]);
if (fire) callbackRef.current();
}, [triggers, triggerType]);
}
@@ -42,7 +42,7 @@ export const defaultOfficeState: OfficeSliceState = {
},
meta: null,
isLoading: false,
refetchTrigger: null,
refetchTriggers: {},
routing: {
byWorkspace: {},
knownProviders: [],
@@ -272,7 +272,8 @@ function createMiscActions(set: SetFn) {
}),
setOfficeRefetchTrigger: (type: string) =>
set((draft) => {
draft.office.refetchTrigger = { type, timestamp: Date.now() };
const prev = draft.office.refetchTriggers[type] ?? 0;
draft.office.refetchTriggers[type] = prev + 1;
}),
};
}
@@ -0,0 +1,32 @@
import type { OfficeTaskPriority, OfficeTaskStatus, TaskLabel } from "./types";
export type OfficeTask = {
id: string;
workspaceId: string;
identifier: string;
title: string;
description?: string;
status: OfficeTaskStatus;
// Pre-normalization backend value (e.g. "SCHEDULING"); read only for a raw
// sub-state the canonical union erases (ExecutionIndicator's "Live" dot).
rawStatus?: string;
priority: OfficeTaskPriority;
parentId?: string;
projectId?: string;
assigneeAgentProfileId?: string;
labels?: TaskLabel[] | string[];
blockedBy?: string[];
children?: OfficeTask[];
executionPolicy?: string;
executionState?: string;
// Derived best-effort from the status-change activity log; not a column.
startedAt?: string;
completedAt?: string;
createdAt: string;
updatedAt: string;
// True when the task lives in a kandev-managed system workflow
// (today: standing coordination; future: routine-fired). The Tasks
// UI renders a "System" badge for these when the dev toggle reveals
// them.
isSystem?: boolean;
};
+8 -33
View File
@@ -254,33 +254,8 @@ export type TaskLabel = {
color: string;
};
export type OfficeTask = {
id: string;
workspaceId: string;
identifier: string;
title: string;
description?: string;
status: OfficeTaskStatus;
// Pre-normalization backend value (e.g. "SCHEDULING"); read only for a raw
// sub-state the canonical union erases (ExecutionIndicator's "Live" dot).
rawStatus?: string;
priority: OfficeTaskPriority;
parentId?: string;
projectId?: string;
assigneeAgentProfileId?: string;
labels?: TaskLabel[] | string[];
blockedBy?: string[];
children?: OfficeTask[];
executionPolicy?: string;
executionState?: string;
createdAt: string;
updatedAt: string;
// True when the task lives in a kandev-managed system workflow
// (today: standing coordination; future: routine-fired). The Tasks
// UI renders a "System" badge for these when the dev toggle reveals
// them.
isSystem?: boolean;
};
import type { OfficeTask } from "./office-task-type";
export type { OfficeTask } from "./office-task-type";
export type TaskFilterState = {
statuses: OfficeTaskStatus[];
@@ -523,11 +498,6 @@ import type {
// --- Slice state & actions ---
export type OfficeRefetchTrigger = {
type: string;
timestamp: number;
};
/**
* Office collections that belong to one workspace, stored per workspace id
* rather than as a single current value.
@@ -562,7 +532,12 @@ export type OfficeSliceState = {
tasks: TasksState;
meta: OfficeMeta | null;
isLoading: boolean;
refetchTrigger: OfficeRefetchTrigger | null;
// Per-type counters rather than one "last trigger" value: a single WS
// handler often fires several distinct types in the same synchronous
// call (e.g. `task:${id}` then `dashboard`), and React/Zustand coalesce
// those into one render, so a shared last-value field would only ever
// let the final type's subscribers see a change. See useOfficeRefetch.
refetchTriggers: Record<string, number>;
routing: RoutingState;
providerHealth: ProviderHealthSliceState;
runAttempts: RunAttemptsState;
+35 -2
View File
@@ -134,7 +134,7 @@ describe("office WS handler — workspace filter", () => {
});
});
describe("office WS handler — task field updates delegate to patchTaskInStore", () => {
describe("office WS handler — task field updates", () => {
beforeEach(() => {
vi.clearAllMocks();
});
@@ -156,7 +156,7 @@ describe("office WS handler — task field updates delegate to patchTaskInStore"
);
});
it("refreshes tasks for a producer task.moved payload without a status", () => {
it("refreshes the task detail and activity for a producer task.moved payload without a status", () => {
const { store, patchTaskInStore, setOfficeRefetchTrigger } = makeStore(ACTIVE_WS);
const handlers = registerOfficeHandlers(store);
const handler = handlers["office.task.moved"]!;
@@ -176,6 +176,8 @@ describe("office WS handler — task field updates delegate to patchTaskInStore"
expect(patchTaskInStore).not.toHaveBeenCalled();
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("tasks");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("dashboard");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("task:t-1");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("activity");
});
it("passes the raw new_status through on office.task.status_changed", () => {
@@ -191,6 +193,37 @@ describe("office WS handler — task field updates delegate to patchTaskInStore"
expect(patchTaskInStore).toHaveBeenCalledWith("t-1", { status: "CREATED" });
});
it("refreshes the per-task DTO and dashboard on office.task.status_changed", () => {
const { store, setOfficeRefetchTrigger } = makeStore(ACTIVE_WS);
const handlers = registerOfficeHandlers(store);
const handler = handlers["office.task.status_changed"]!;
handler({
type: "notification",
action: "office.task.status_changed",
payload: { workspace_id: ACTIVE_WS, task_id: "t-42", new_status: "done" },
} as Parameters<typeof handler>[0]);
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("task:t-42");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("dashboard");
});
it("refreshes the per-task DTO, dashboard, and activity on office.task.moved", () => {
const { store, setOfficeRefetchTrigger } = makeStore(ACTIVE_WS);
const handlers = registerOfficeHandlers(store);
const handler = handlers["office.task.moved"]!;
handler({
type: "notification",
action: "office.task.moved",
payload: { workspace_id: ACTIVE_WS, task_id: "t-99", new_status: "in_progress" },
} as Parameters<typeof handler>[0]);
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("task:t-99");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("dashboard");
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("activity");
});
});
describe("office WS handler — approval flow events", () => {
+11 -2
View File
@@ -79,12 +79,20 @@ function buildTaskHandlers(
if (!isCurrentWorkspace(p)) return;
const taskId = (p.task_id ?? p.id) as string | undefined;
const newStatus = p.new_status as string | undefined;
if (!taskId || !newStatus) {
if (!taskId) {
triggerRefetch("tasks");
triggerRefetch("dashboard");
return;
}
updateTaskStatus(taskId, { status: newStatus as OfficeTaskStatus });
// The task service publishes task.moved without new_status. Keep the
// direct patch when a producer provides one, but always refresh the
// detail and activity projections for a known task.
if (newStatus) {
updateTaskStatus(taskId, { status: newStatus as OfficeTaskStatus });
} else {
triggerRefetch("tasks");
}
triggerRefetch(`task:${taskId}`);
triggerRefetch("dashboard");
triggerRefetch("activity");
},
@@ -99,6 +107,7 @@ function buildTaskHandlers(
return;
}
updateTaskStatus(taskId, { status: newStatus as OfficeTaskStatus });
triggerRefetch(`task:${taskId}`);
triggerRefetch("dashboard");
},