fix(office): make repeat task reassignment work again after 24 hours

apps/backend/internal/backendapp/adapters_office.go

@@ -17,6 +17,7 @@ import (
 	officeroutines "github.com/kandev/kandev/internal/office/routines"
 	officeservice "github.com/kandev/kandev/internal/office/service"
 	officewakeup "github.com/kandev/kandev/internal/office/wakeup"
+	runsservice "github.com/kandev/kandev/internal/runs/service"
 	"github.com/kandev/kandev/internal/task/models"
 	tasksqlite "github.com/kandev/kandev/internal/task/repository/sqlite"
 	taskservice "github.com/kandev/kandev/internal/task/service"
@@ -266,7 +267,12 @@ func (a *routineWakeupAdapter) CreateWakeupRequest(
 	}
 	if err := a.repo.CreateWakeupRequest(ctx, row); err != nil {
 		if errors.Is(err, officesqlite.ErrWakeupIdempotencyConflict) {
-			return officeroutines.ErrWakeupAlreadyRequested
+			runsservice.ReportDurableDedup(runsservice.QueueSourceWakeup, req.Reason, req.IdempotencyKey, req.AgentProfileID)
+			// Wraps both sentinels so a caller can check either: routines
+			// callers key off ErrWakeupAlreadyRequested to treat this as
+			// success by another route, while errors.Is against the
+			// sqlite-layer ErrWakeupIdempotencyConflict still matches.
+			return fmt.Errorf("%w: %w", officeroutines.ErrWakeupAlreadyRequested, err)
 		}
 		return err
 	}

apps/backend/internal/backendapp/adapters_office_wakeup_test.go

@@ -0,0 +1,97 @@
+package backendapp
+
+import (
+	"context"
+	"errors"
+	"expvar"
+	"strings"
+	"testing"
+
+	"github.com/jmoiron/sqlx"
+	_ "github.com/mattn/go-sqlite3"
+
+	settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
+	officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
+	officeroutines "github.com/kandev/kandev/internal/office/routines"
+)
+
+func newTestRoutineWakeupAdapter(t *testing.T) *routineWakeupAdapter {
+	t.Helper()
+	db, err := sqlx.Open("sqlite3", ":memory:")
+	if err != nil {
+		t.Fatalf("open sqlite: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	if _, _, err := settingsstore.Provide(db, db, nil); err != nil {
+		t.Fatalf("settings store init: %v", err)
+	}
+	repo, err := officesqlite.NewWithDB(db, db, nil)
+	if err != nil {
+		t.Fatalf("new office repo: %v", err)
+	}
+	return &routineWakeupAdapter{repo: repo}
+}
+
+// AC-OFFICE-RUN-DEDUP-004.6: a durable idempotency conflict on the wakeup
+// queue moves office_run_dedup_total{queue="wakeup",kind="durable"} through
+// ReportDurableDedup, not ReportInsertResult - the adapter does its own
+// errors.Is on the repository's sentinel and reports the conflict itself.
+func TestRoutineWakeupAdapter_CreateWakeupRequest_DurableConflictReportsCounter(t *testing.T) {
+	adapter := newTestRoutineWakeupAdapter(t)
+	ctx := context.Background()
+	reason := "test_routine_wakeup_conflict_" + t.Name()
+	key := "routine:r1:t1:tick:1234567890"
+
+	first := &officeroutines.WakeupRequest{
+		ID: "wakeup-1", AgentProfileID: "agent-1", Source: "cron",
+		Reason: reason, IdempotencyKey: key,
+	}
+	if err := adapter.CreateWakeupRequest(ctx, first); err != nil {
+		t.Fatalf("first create: %v", err)
+	}
+
+	second := &officeroutines.WakeupRequest{
+		ID: "wakeup-2", AgentProfileID: "agent-1", Source: "cron",
+		Reason: reason, IdempotencyKey: key,
+	}
+	err := adapter.CreateWakeupRequest(ctx, second)
+	if !errors.Is(err, officesqlite.ErrWakeupIdempotencyConflict) {
+		t.Fatalf("second create error = %v, want ErrWakeupIdempotencyConflict", err)
+	}
+
+	// reason isn't in runs/service's bounded metricReasons allowlist, so the
+	// label buckets it to "custom" rather than carrying it verbatim.
+	if !counterHasLabel(t, "office_run_dedup_total", "reason=custom", "kind=durable", "queue=wakeup") {
+		t.Fatal("expected office_run_dedup_total to carry a custom/durable/wakeup entry")
+	}
+}
+
+// counterHasLabel reports whether the named expvar.Map has an entry whose
+// key contains every given substring. Counters are process-global expvar
+// state (no reset hook), so tests assert presence/growth rather than exact
+// values that could collide with other tests in the same binary run.
+func counterHasLabel(t *testing.T, mapName string, substrs ...string) bool {
+	t.Helper()
+	v := expvar.Get(mapName)
+	if v == nil {
+		t.Fatalf("expvar map %q not registered", mapName)
+	}
+	m, ok := v.(*expvar.Map)
+	if !ok {
+		t.Fatalf("expvar %q is not a *expvar.Map", mapName)
+	}
+	found := false
+	m.Do(func(kv expvar.KeyValue) {
+		matches := true
+		for _, s := range substrs {
+			if !strings.Contains(kv.Key, s) {
+				matches = false
+				break
+			}
+		}
+		if matches {
+			found = true
+		}
+	})
+	return found
+}

apps/backend/internal/office/approvals/handler_security_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/kandev/kandev/internal/office/models"
 	"github.com/kandev/kandev/internal/office/repository/sqlite"
 	"github.com/kandev/kandev/internal/office/shared"
+	runsservice "github.com/kandev/kandev/internal/runs/service"
 )
 
 // approvalHandlerFixture wires the minimal stack needed to exercise the
@@ -72,7 +73,9 @@ func (s *silentActivityLogger) LogActivityWithRun(_ context.Context, _, _, _, _,
 
 type silentRunQueuer struct{}
 
-func (s *silentRunQueuer) QueueRun(_ context.Context, _, _, _, _ string) error { return nil }
+func (s *silentRunQueuer) QueueRun(_ context.Context, _, _, _, _ string) (runsservice.QueueOutcome, error) {
+	return runsservice.QueueOutcomeQueued, nil
+}
 
 // seedApprovalAgent creates an agent_profiles row with the given role
 // and permissions, returning the persisted instance.

apps/backend/internal/office/approvals/service.go

@@ -9,6 +9,7 @@ import (
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/office/models"
 	"github.com/kandev/kandev/internal/office/shared"
+	runsservice "github.com/kandev/kandev/internal/runs/service"
 
 	"go.uber.org/zap"
 )
@@ -29,7 +30,7 @@ type AgentWriter interface {
 
 // RunQueuer enqueues run requests for agent instances.
 type RunQueuer interface {
-	QueueRun(ctx context.Context, agentInstanceID, reason, payload, idempotencyKey string) error
+	QueueRun(ctx context.Context, agentInstanceID, reason, payload, idempotencyKey string) (runsservice.QueueOutcome, error)
 }
 
 // ApprovalService handles approval CRUD and decide logic.
@@ -215,6 +216,7 @@ func (s *ApprovalService) queueApprovalRun(ctx context.Context, approval *Approv
 		approval.ID, approval.Type, approval.Status, approval.DecisionNote,
 	)
 	idempotencyKey := "approval:" + approval.ID
-	return s.runs.QueueRun(ctx, approval.RequestedByAgentProfileID,
+	_, err := s.runs.QueueRun(ctx, approval.RequestedByAgentProfileID,
 		"approval_resolved", payload, idempotencyKey)
+	return err
 }

apps/backend/internal/office/approvals/service_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/kandev/kandev/internal/office/approvals"
 	"github.com/kandev/kandev/internal/office/models"
 	"github.com/kandev/kandev/internal/office/repository/sqlite"
+	runsservice "github.com/kandev/kandev/internal/runs/service"
 )
 
 // noopActivityLogger implements shared.ActivityLogger without importing shared.
@@ -23,7 +24,9 @@ func (n *noopActivityLogger) LogActivityWithRun(_ context.Context, _, _, _, _, _
 // noopRunQueuer implements approvals.RunQueuer as a no-op.
 type noopRunQueuer struct{}
 
-func (n *noopRunQueuer) QueueRun(_ context.Context, _, _, _, _ string) error { return nil }
+func (n *noopRunQueuer) QueueRun(_ context.Context, _, _, _, _ string) (runsservice.QueueOutcome, error) {
+	return runsservice.QueueOutcomeQueued, nil
+}
 
 type fakeAgentWriter struct {
 	statuses map[string]string

apps/backend/internal/office/channels/service.go

@@ -103,7 +103,7 @@ func (s *ChannelService) SetupChannel(ctx context.Context, channel *models.Chann
 	// workflow_step_participants. Channel tasks have no workflow_step_id;
 	// the participant row is keyed at (step_id="", task_id) and the
 	// runner projection still resolves it.
-	if err := s.repo.UpdateTaskAssignee(ctx, taskID, channel.AgentProfileID); err != nil {
+	if _, err := s.repo.UpdateTaskAssignee(ctx, taskID, channel.AgentProfileID); err != nil {
 		_ = s.repo.DeleteChannel(ctx, channel.ID)
 		return fmt.Errorf("set channel task assignee: %w", err)
 	}

apps/backend/internal/office/dashboard/handler_test.go

@@ -141,6 +141,7 @@ func newTestDeps(t *testing.T) *testDeps {
 			project_id TEXT DEFAULT '',
 			assignee_agent_profile_id TEXT DEFAULT '',
 			assignee_user_id TEXT NOT NULL DEFAULT '',
+			assignment_generation INTEGER NOT NULL DEFAULT 0,
 			labels TEXT DEFAULT '[]',
 			metadata TEXT DEFAULT '{}',
 			identifier TEXT DEFAULT '',

apps/backend/internal/office/dashboard/service.go

@@ -73,8 +73,11 @@ type Repository interface {
 	GetRunsByCommentIDs(ctx context.Context, commentIDs []string) (map[string]sqlite.CommentRunStatus, error)
 	UpdateTaskState(ctx context.Context, taskID, state string) error
 	GetTaskExecutionFields(ctx context.Context, taskID string) (*sqlite.TaskExecutionFields, error)
+	// UpdateTaskAssignee returns the task's assignment_generation after the
+	// bump, read back inside the same transaction that wrote the runner
+	// seat (see the sqlite implementation's doc comment).
+	UpdateTaskAssignee(ctx context.Context, taskID, assigneeID string) (int64, error)
 	UpdateTaskStateIfWorkflowStep(ctx context.Context, taskID, expectedStepID, state string) (bool, error)
-	UpdateTaskAssignee(ctx context.Context, taskID, assigneeID string) error
 	UpdateTaskPriority(ctx context.Context, taskID, priority string) error
 	UpdateTaskProjectID(ctx context.Context, taskID, projectID string) error
 	GetTaskProjectID(ctx context.Context, taskID string) (string, error)
@@ -254,6 +257,11 @@ type MarkFixedHandler interface {
 type TaskReactivityChange struct {
 	NewStatus     *string
 	NewAssigneeID *string
+	// AssignmentGeneration is the value UpdateTaskAssignee's transaction
+	// committed and read back, carried here rather than re-read. Nil means
+	// the caller could not supply one (e.g. the read-back itself failed);
+	// the pipeline then enqueues the task_assigned wake keyless.
+	AssignmentGeneration *int64
 	// PrevAssigneeID is the assignee BEFORE the mutation. Required when
 	// NewAssigneeID is set so the pipeline can detect a real change and
 	// hand off the previous assignee's session.

apps/backend/internal/office/dashboard/service_tasks.go

@@ -915,22 +915,25 @@ func (s *DashboardService) SetTaskAssigneeAsAgent(ctx context.Context, callerAge
 				zap.String("task_id", taskID), zap.Error(err))
 		}
 	}
-	if err := s.repo.UpdateTaskAssignee(ctx, taskID, assigneeID); err != nil {
+	generation, err := s.repo.UpdateTaskAssignee(ctx, taskID, assigneeID)
+	if err != nil {
 		return err
 	}
 
 	s.publishTaskUpdated(ctx, taskID, []string{"assignee_agent_profile_id"})
 
 	// Reactivity pipeline — wakes the new assignee with task_assigned
 	// and hard-cancels the previous assignee's active session.
-	s.runReactivityForAssigneeChange(ctx, taskID, prevAssignee, assigneeID, callerAgentID)
+	s.runReactivityForAssigneeChange(ctx, taskID, prevAssignee, assigneeID, callerAgentID, generation)
 	return nil
 }
 
 // runReactivityForAssigneeChange invokes the reactivity pipeline for an
 // assignee change. Best-effort — failures are logged, never propagated.
+// generation is the value UpdateTaskAssignee's transaction just committed
+// and read back; it is carried onto the mutation rather than re-read.
 func (s *DashboardService) runReactivityForAssigneeChange(
-	ctx context.Context, taskID, prevAssigneeID, newAssigneeID, callerAgentID string,
+	ctx context.Context, taskID, prevAssigneeID, newAssigneeID, callerAgentID string, generation int64,
 ) {
 	if s.reactivity == nil {
 		return
@@ -940,10 +943,11 @@ func (s *DashboardService) runReactivityForAssigneeChange(
 		actorType = "agent"
 	}
 	change := TaskReactivityChange{
-		NewAssigneeID:  &newAssigneeID,
-		PrevAssigneeID: prevAssigneeID,
-		ActorID:        callerAgentID,
-		ActorType:      actorType,
+		NewAssigneeID:        &newAssigneeID,
+		AssignmentGeneration: &generation,
+		PrevAssigneeID:       prevAssigneeID,
+		ActorID:              callerAgentID,
+		ActorType:            actorType,
 	}
 	// preStatus="" — assignee changes don't depend on the prev status.
 	result, err := s.reactivity.ApplyTaskMutation(ctx, taskID, "", change)
@@ -958,7 +962,10 @@ func (s *DashboardService) runReactivityForAssigneeChange(
 	// Flip the prev assignee's office session row to COMPLETED so it leaves
 	// the active sessions list. The reactivity pipeline already hard-cancels
 	// the running execution above; this is the persistent-row counterpart.
-	if prevAssigneeID != "" && s.sessionTerm != nil {
+	// A same-agent reassignment is not a handoff — the pipeline above never
+	// interrupts it — so this must not terminate the agent's own live
+	// session out from under it.
+	if prevAssigneeID != "" && prevAssigneeID != newAssigneeID && s.sessionTerm != nil {
 		if err := s.sessionTerm.TerminateOfficeSession(ctx, taskID, prevAssigneeID, sessionTermReasonReassigned); err != nil {
 			s.logger.Warn("terminate prev-assignee office session failed",
 				zap.String("task_id", taskID),
@@ -970,7 +977,7 @@ func (s *DashboardService) runReactivityForAssigneeChange(
 	// the user isn't asked to triage a failure they already worked
 	// around by reassigning. Counter is intentionally not reset — the
 	// root cause may still be unfixed for the old agent.
-	if prevAssigneeID != "" && s.failureNotifier != nil {
+	if prevAssigneeID != "" && prevAssigneeID != newAssigneeID && s.failureNotifier != nil {
 		s.failureNotifier.OnAssigneeChanged(ctx, taskID, prevAssigneeID)
 	}
 }

apps/backend/internal/office/dashboard/session_termination_test.go

@@ -13,6 +13,18 @@ type recordingTerminator struct {
 	calls []termCall
 }
 
+type recordingFailureNotifier struct {
+	calls []struct {
+		taskID, agentID string
+	}
+}
+
+func (r *recordingFailureNotifier) OnAssigneeChanged(_ context.Context, taskID, agentID string) {
+	r.calls = append(r.calls, struct {
+		taskID, agentID string
+	}{taskID: taskID, agentID: agentID})
+}
+
 type termCall struct {
 	taskID, agentID, reason string
 }
@@ -85,12 +97,14 @@ func (r *recordingReactivity) ApplyTaskMutation(_ context.Context, _ string, _ s
 func TestSetTaskAssignee_TerminatesPrevSession(t *testing.T) {
 	deps := newTestDeps(t)
 	rt := &recordingTerminator{}
+	fn := &recordingFailureNotifier{}
 	deps.svc.SetSessionTerminator(rt)
+	deps.svc.SetFailureNotifier(fn)
 	deps.svc.SetReactivityApplier(&recordingReactivity{result: &dashboard.TaskReactivityResult{}})
 
 	insertTestTask(t, deps.db, "task-r", "ws-r", "Reassign", "todo", 2)
 	// Seed prev assignee directly via the underlying repo update.
-	if err := deps.repo.UpdateTaskAssignee(context.Background(), "task-r", "agent-prev"); err != nil {
+	if _, err := deps.repo.UpdateTaskAssignee(context.Background(), "task-r", "agent-prev"); err != nil {
 		t.Fatalf("seed prev assignee: %v", err)
 	}
 
@@ -105,4 +119,38 @@ func TestSetTaskAssignee_TerminatesPrevSession(t *testing.T) {
 	if got.taskID != "task-r" || got.agentID != "agent-prev" {
 		t.Errorf("term call: got %+v", got)
 	}
+	if len(fn.calls) != 1 || fn.calls[0].taskID != "task-r" || fn.calls[0].agentID != "agent-prev" {
+		t.Fatalf("expected prior-assignee failure notification, got %+v", fn.calls)
+	}
+}
+
+// TestSetTaskAssignee_SameAgent_DoesNotTerminateSession is the regression
+// test for Review round 3 Finding 5: a repeat assignment to the agent that
+// already holds the seat must not flip that agent's own live session row to
+// COMPLETED. The reactivity pipeline correctly declines to hard-cancel this
+// case (AC-OFFICE-RUN-DEDUP-001.3 treats it as a real occurrence, not an
+// interrupt), and the persisted-row side effect must agree.
+func TestSetTaskAssignee_SameAgent_DoesNotTerminateSession(t *testing.T) {
+	deps := newTestDeps(t)
+	rt := &recordingTerminator{}
+	fn := &recordingFailureNotifier{}
+	deps.svc.SetSessionTerminator(rt)
+	deps.svc.SetFailureNotifier(fn)
+	deps.svc.SetReactivityApplier(&recordingReactivity{result: &dashboard.TaskReactivityResult{}})
+
+	insertTestTask(t, deps.db, "task-same", "ws-r", "Reassign", "todo", 2)
+	if _, err := deps.repo.UpdateTaskAssignee(context.Background(), "task-same", "agent-x"); err != nil {
+		t.Fatalf("seed prev assignee: %v", err)
+	}
+
+	if err := deps.svc.SetTaskAssigneeAsAgent(context.Background(), "", "task-same", "agent-x"); err != nil {
+		t.Fatalf("set assignee: %v", err)
+	}
+
+	if len(rt.calls) != 0 {
+		t.Fatalf("same-agent reassignment must not terminate the agent's own session, got %d calls (%+v)", len(rt.calls), rt.calls)
+	}
+	if len(fn.calls) != 0 {
+		t.Fatalf("same-agent reassignment must not dismiss its failure inbox entry, got %+v", fn.calls)
+	}
 }

apps/backend/internal/office/onboarding/service.go

@@ -17,6 +17,7 @@ import (
 	"github.com/kandev/kandev/internal/office/repository/sqlite"
 	"github.com/kandev/kandev/internal/office/routing"
 	"github.com/kandev/kandev/internal/office/shared"
+	runsservice "github.com/kandev/kandev/internal/runs/service"
 	taskservice "github.com/kandev/kandev/internal/task/service"
 
 	"go.uber.org/zap"
@@ -349,7 +350,11 @@ func (s *OnboardingService) maybeCreateOnboardingTask(
 		return ""
 	}
 	if s.runQueuer != nil {
-		if wakeErr := s.runQueuer.QueueRun(ctx, agentID, runReasonTaskAssigned,
+		// A third task_assigned producer alongside queueTaskAssignedRun; it is
+		// never handed the assigning transaction's generation, so it enqueues
+		// keyless rather than deriving a divergent key.
+		runsservice.ReportKeylessEnqueue(runReasonTaskAssigned, runsservice.KeylessCauseUnresolved, "onboarding_no_generation")
+		if _, wakeErr := s.runQueuer.QueueRun(ctx, agentID, runReasonTaskAssigned,
 			fmt.Sprintf(`{"task_id":%q}`, taskID), ""); wakeErr != nil {
 			s.logger.Warn("enqueue onboarding run failed", zap.Error(wakeErr))
 		}