fix(backend): fire budget alerts when reassigning a task pushes a project over budget (#3276)
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -2085,6 +2085,7 @@ func buildOfficeDashboardService(
|
||||
cfgWriter *configloader.FileWriter,
|
||||
) *officedashboard.DashboardService {
|
||||
dashboardSvc := officedashboard.NewDashboardService(repo, log, activity, agentSvc, costSvc)
|
||||
dashboardSvc.SetProjectBudgetEvaluator(costSvc)
|
||||
dashboardSvc.SetGovernanceStore(repo)
|
||||
dashboardSvc.SetSkillLister(skillSvc)
|
||||
dashboardSvc.SetRoutineLister(routineSvc)
|
||||
|
||||
@@ -214,6 +214,40 @@ func (s *CostService) CheckPreExecutionBudget(
|
||||
return true, "", nil
|
||||
}
|
||||
|
||||
// EvaluateProjectBudget evaluates project-scoped budget policies for
|
||||
// projectID after a task is reassigned into it — the only budget hook on
|
||||
// the reassignment path. Reassignment can move a task's historical spend
|
||||
// across the project boundary that GetCostForProjectSince rolls up, so
|
||||
// without this the destination project's policies never see the crossing.
|
||||
//
|
||||
// Only policies with ScopeType==project and ScopeID==projectID are
|
||||
// evaluated: reassignment does not change the workspace total, so
|
||||
// re-checking scopeWorkspace policies (as CheckBudget does) would emit a
|
||||
// duplicate alert on every reassignment in an already over-budget
|
||||
// workspace. The source project is not evaluated either — reassignment
|
||||
// only lowers its spend, so it can't newly cross a threshold. A no-op
|
||||
// projectID (clearing a project) has no destination to evaluate.
|
||||
func (s *CostService) EvaluateProjectBudget(ctx context.Context, workspaceID, projectID string) error {
|
||||
if projectID == "" {
|
||||
return nil
|
||||
}
|
||||
policies, err := s.repo.ListBudgetPolicies(ctx, workspaceID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, policy := range policies {
|
||||
if policy.ScopeType != scopeProject || policy.ScopeID != projectID {
|
||||
continue
|
||||
}
|
||||
if _, err := s.evaluatePolicy(ctx, workspaceID, policy); err != nil {
|
||||
s.logger.Error("project budget check failed",
|
||||
zap.String("policy_id", policy.ID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *CostService) pauseAgentForBudget(ctx context.Context, agentID string) bool {
|
||||
agent, err := s.agents.GetAgentInstance(ctx, agentID)
|
||||
if err != nil {
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/office/costs"
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/repository/sqlite"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
)
|
||||
|
||||
// repoAgents adapts the office repository to shared.AgentReader +
|
||||
@@ -40,6 +41,16 @@ func (a *repoAgents) UpdateAgentStatusFields(ctx context.Context, agentID, statu
|
||||
}
|
||||
|
||||
func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
|
||||
t.Helper()
|
||||
return newBudgetTestServiceWithActivity(t, &noopActivity{})
|
||||
}
|
||||
|
||||
// newBudgetTestServiceWithActivity is the same setup as newBudgetTestService
|
||||
// but lets project-scope tests supply a spy activity logger to observe which
|
||||
// budget.alert / budget.exceeded rows fire.
|
||||
func newBudgetTestServiceWithActivity(
|
||||
t *testing.T, activity shared.ActivityLogger,
|
||||
) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
|
||||
t.Helper()
|
||||
db, err := sqlx.Open("sqlite3", ":memory:")
|
||||
if err != nil {
|
||||
@@ -54,6 +65,7 @@ func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository,
|
||||
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS tasks (
|
||||
id TEXT PRIMARY KEY,
|
||||
workspace_id TEXT NOT NULL DEFAULT '',
|
||||
project_id TEXT DEFAULT '',
|
||||
state TEXT NOT NULL DEFAULT 'TODO',
|
||||
title TEXT DEFAULT '',
|
||||
description TEXT DEFAULT '',
|
||||
@@ -70,7 +82,7 @@ func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository,
|
||||
}
|
||||
log := logger.Default()
|
||||
agents := &repoAgents{repo: repo}
|
||||
svc := costs.NewCostService(repo, log, &noopActivity{}, agents, agents)
|
||||
svc := costs.NewCostService(repo, log, activity, agents, agents)
|
||||
|
||||
execSQL := func(query string, args ...interface{}) {
|
||||
t.Helper()
|
||||
@@ -81,6 +93,41 @@ func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository,
|
||||
return svc, repo, execSQL
|
||||
}
|
||||
|
||||
// insertBudgetTestTask inserts a minimal task row carrying a project_id, so
|
||||
// project-scoped cost rollups (which join office_cost_events to tasks on
|
||||
// project_id) resolve. Agent-scoped tests don't need this: GetCostForAgentSince
|
||||
// reads office_cost_events directly with no join.
|
||||
func insertBudgetTestTask(t *testing.T, execSQL func(string, ...interface{}), taskID, workspaceID, projectID string) {
|
||||
t.Helper()
|
||||
execSQL(
|
||||
`INSERT INTO tasks (id, workspace_id, project_id) VALUES (?, ?, ?)`,
|
||||
taskID, workspaceID, projectID,
|
||||
)
|
||||
}
|
||||
|
||||
// budgetActivityCall records one LogActivity invocation observed by
|
||||
// budgetActivitySpy.
|
||||
type budgetActivityCall struct {
|
||||
action string
|
||||
targetType string
|
||||
targetID string
|
||||
}
|
||||
|
||||
// budgetActivitySpy implements shared.ActivityLogger and records every call,
|
||||
// so project-budget tests can assert exactly which alerts fired without
|
||||
// depending on evaluatePolicy's return value (EvaluateProjectBudget only
|
||||
// returns an error).
|
||||
type budgetActivitySpy struct {
|
||||
calls []budgetActivityCall
|
||||
}
|
||||
|
||||
func (s *budgetActivitySpy) LogActivity(_ context.Context, _, _, _, action, targetType, targetID, _ string) {
|
||||
s.calls = append(s.calls, budgetActivityCall{action: action, targetType: targetType, targetID: targetID})
|
||||
}
|
||||
|
||||
func (s *budgetActivitySpy) LogActivityWithRun(_ context.Context, _, _, _, _, _, _, _, _, _ string) {
|
||||
}
|
||||
|
||||
func createBudgetTestAgent(t *testing.T, repo *sqlite.Repository, wsID, agentID string) {
|
||||
t.Helper()
|
||||
agent := &models.AgentInstance{
|
||||
@@ -336,3 +383,179 @@ func TestCheckPreExecutionBudget_PauseAgentBlocks(t *testing.T) {
|
||||
t.Error("pause_agent exceedence must block")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvaluateProjectBudget_AlertAtThreshold covers the reassignment defect:
|
||||
// a project-scoped notify_only policy must fire budget.alert when evaluated
|
||||
// directly, without waiting for a future cost event or agent launch.
|
||||
func TestEvaluateProjectBudget_AlertAtThreshold(t *testing.T) {
|
||||
spy := &budgetActivitySpy{}
|
||||
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
|
||||
ctx := context.Background()
|
||||
|
||||
policy := &models.BudgetPolicy{
|
||||
WorkspaceID: "ws-1",
|
||||
ScopeType: "project",
|
||||
ScopeID: "proj-1",
|
||||
LimitSubcents: 1000,
|
||||
Period: "monthly",
|
||||
AlertThresholdPct: 80,
|
||||
ActionOnExceed: "notify_only",
|
||||
}
|
||||
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
|
||||
t.Fatalf("create policy: %v", err)
|
||||
}
|
||||
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
|
||||
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(850))
|
||||
|
||||
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
|
||||
t.Fatalf("EvaluateProjectBudget: %v", err)
|
||||
}
|
||||
|
||||
if !hasBudgetActivity(spy.calls, "budget.alert", "proj-1") {
|
||||
t.Errorf("expected budget.alert for proj-1, got calls=%+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvaluateProjectBudget_ExceededAtLimit covers the limit-exceeded branch.
|
||||
func TestEvaluateProjectBudget_ExceededAtLimit(t *testing.T) {
|
||||
spy := &budgetActivitySpy{}
|
||||
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
|
||||
ctx := context.Background()
|
||||
|
||||
policy := &models.BudgetPolicy{
|
||||
WorkspaceID: "ws-1",
|
||||
ScopeType: "project",
|
||||
ScopeID: "proj-1",
|
||||
LimitSubcents: 500,
|
||||
Period: "monthly",
|
||||
AlertThresholdPct: 80,
|
||||
ActionOnExceed: "notify_only",
|
||||
}
|
||||
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
|
||||
t.Fatalf("create policy: %v", err)
|
||||
}
|
||||
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
|
||||
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(600))
|
||||
|
||||
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
|
||||
t.Fatalf("EvaluateProjectBudget: %v", err)
|
||||
}
|
||||
|
||||
if !hasBudgetActivity(spy.calls, "budget.exceeded", "proj-1") {
|
||||
t.Errorf("expected budget.exceeded for proj-1, got calls=%+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvaluateProjectBudget_PauseAgentPolicyDoesNotPause locks in the design
|
||||
// decision: a project-scoped pause_agent policy never pauses an agent.
|
||||
// evaluatePolicy only pauses when ScopeType==agent, so this reuses that
|
||||
// existing gate rather than adding new pause logic for project scope. The
|
||||
// agent instance shares its ID with the policy's ScopeID so a regression
|
||||
// that mistakenly treated ScopeID as an agent ID would find and pause it.
|
||||
func TestEvaluateProjectBudget_PauseAgentPolicyDoesNotPause(t *testing.T) {
|
||||
spy := &budgetActivitySpy{}
|
||||
svc, repo, execSQL := newBudgetTestServiceWithActivity(t, spy)
|
||||
ctx := context.Background()
|
||||
|
||||
createBudgetTestAgent(t, repo, "ws-1", "proj-1")
|
||||
|
||||
policy := &models.BudgetPolicy{
|
||||
WorkspaceID: "ws-1",
|
||||
ScopeType: "project",
|
||||
ScopeID: "proj-1",
|
||||
LimitSubcents: 500,
|
||||
Period: "monthly",
|
||||
AlertThresholdPct: 80,
|
||||
ActionOnExceed: "pause_agent",
|
||||
}
|
||||
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
|
||||
t.Fatalf("create policy: %v", err)
|
||||
}
|
||||
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
|
||||
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(600))
|
||||
|
||||
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
|
||||
t.Fatalf("EvaluateProjectBudget: %v", err)
|
||||
}
|
||||
|
||||
agent, err := repo.GetAgentInstance(ctx, "proj-1")
|
||||
if err != nil {
|
||||
t.Fatalf("get agent: %v", err)
|
||||
}
|
||||
if agent.Status == "paused" {
|
||||
t.Error("project-scoped pause_agent policy must not pause an agent")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvaluateProjectBudget_WorkspacePoliciesNotEvaluated locks in the
|
||||
// decision to skip scope=workspace policies: reassignment doesn't change the
|
||||
// workspace total, so re-evaluating them would emit a duplicate alert row on
|
||||
// every reassignment in an already over-budget workspace.
|
||||
func TestEvaluateProjectBudget_WorkspacePoliciesNotEvaluated(t *testing.T) {
|
||||
spy := &budgetActivitySpy{}
|
||||
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
|
||||
ctx := context.Background()
|
||||
|
||||
policy := &models.BudgetPolicy{
|
||||
WorkspaceID: "ws-1",
|
||||
ScopeType: "workspace",
|
||||
ScopeID: "",
|
||||
LimitSubcents: 500,
|
||||
Period: "monthly",
|
||||
AlertThresholdPct: 80,
|
||||
ActionOnExceed: "notify_only",
|
||||
}
|
||||
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
|
||||
t.Fatalf("create policy: %v", err)
|
||||
}
|
||||
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
|
||||
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(600))
|
||||
|
||||
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
|
||||
t.Fatalf("EvaluateProjectBudget: %v", err)
|
||||
}
|
||||
|
||||
if len(spy.calls) != 0 {
|
||||
t.Errorf("expected no activity for workspace-scoped policies, got calls=%+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvaluateProjectBudget_EmptyProjectIDSkips covers clearing a project
|
||||
// (projectID=="") — there is no destination to evaluate.
|
||||
func TestEvaluateProjectBudget_EmptyProjectIDSkips(t *testing.T) {
|
||||
spy := &budgetActivitySpy{}
|
||||
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
|
||||
ctx := context.Background()
|
||||
|
||||
policy := &models.BudgetPolicy{
|
||||
WorkspaceID: "ws-1",
|
||||
ScopeType: "project",
|
||||
ScopeID: "proj-1",
|
||||
LimitSubcents: 500,
|
||||
Period: "monthly",
|
||||
AlertThresholdPct: 80,
|
||||
ActionOnExceed: "notify_only",
|
||||
}
|
||||
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
|
||||
t.Fatalf("create policy: %v", err)
|
||||
}
|
||||
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
|
||||
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(600))
|
||||
|
||||
if err := svc.EvaluateProjectBudget(ctx, "ws-1", ""); err != nil {
|
||||
t.Fatalf("EvaluateProjectBudget: %v", err)
|
||||
}
|
||||
|
||||
if len(spy.calls) != 0 {
|
||||
t.Errorf("expected no activity when projectID is empty, got calls=%+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func hasBudgetActivity(calls []budgetActivityCall, action, targetID string) bool {
|
||||
for _, c := range calls {
|
||||
if c.action == action && c.targetID == targetID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package dashboard_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -8,9 +9,27 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/office/dashboard"
|
||||
"github.com/kandev/kandev/internal/office/models"
|
||||
"github.com/kandev/kandev/internal/office/shared"
|
||||
)
|
||||
|
||||
func insertBudgetInboxActivity(t *testing.T, deps *testDeps, action string) {
|
||||
t.Helper()
|
||||
if err := deps.repo.CreateActivityEntry(context.Background(), &models.ActivityEntry{
|
||||
ID: "inbox-" + action,
|
||||
WorkspaceID: "ws1",
|
||||
ActorType: models.ActivityActorSystem,
|
||||
ActorID: "budget_checker",
|
||||
Action: models.ActivityAction(action),
|
||||
TargetType: models.ActivityTargetProject,
|
||||
TargetID: "project-1",
|
||||
Details: "budget details",
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}); err != nil {
|
||||
t.Fatalf("insert %s activity: %v", action, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetInbox_IncludesPermissionRequestItems(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
|
||||
@@ -83,3 +102,55 @@ func TestGetInboxCount_IncludesPermissionRequests(t *testing.T) {
|
||||
t.Errorf("expected count >= 2 (including 2 permission requests), got %d", resp.Count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetInbox_IncludesBudgetExceededActivity(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertBudgetInboxActivity(t, deps, "budget.exceeded")
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/office/workspaces/ws1/inbox", 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.InboxResponse
|
||||
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
for _, item := range resp.Items {
|
||||
if item.ID == "inbox-budget.exceeded" {
|
||||
if item.Type != "budget_alert" {
|
||||
t.Errorf("type = %q, want budget_alert", item.Type)
|
||||
}
|
||||
if item.Title != "Budget exceeded" {
|
||||
t.Errorf("title = %q, want Budget exceeded", item.Title)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatal("expected budget.exceeded activity in inbox")
|
||||
}
|
||||
|
||||
func TestGetInboxCount_IncludesBudgetExceededActivity(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertBudgetInboxActivity(t, deps, "budget.exceeded")
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/office/workspaces/ws1/inbox?count=true", 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.InboxCountResponse
|
||||
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if resp.Count < 1 {
|
||||
t.Fatalf("expected count >= 1 for budget.exceeded activity, got %d", resp.Count)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -76,6 +76,7 @@ type Repository interface {
|
||||
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)
|
||||
UpdateTaskParentID(ctx context.Context, taskID, parentID string) error
|
||||
GetProjectWorkspaceID(ctx context.Context, projectID string) (string, error)
|
||||
GetTaskWorkspaceID(ctx context.Context, taskID string) (string, error)
|
||||
@@ -145,6 +146,15 @@ type RetryCanceller interface {
|
||||
CancelPendingRetriesForTask(ctx context.Context, taskID string) error
|
||||
}
|
||||
|
||||
// ProjectBudgetEvaluator evaluates project-scoped budget policies for a
|
||||
// destination project. Wired to costs.CostService.EvaluateProjectBudget so
|
||||
// reassigning a task into a project re-checks that project's policies —
|
||||
// otherwise a notify_only threshold crossed purely by moving historical
|
||||
// spend (not a new cost event or agent launch) would never fire an alert.
|
||||
type ProjectBudgetEvaluator interface {
|
||||
EvaluateProjectBudget(ctx context.Context, workspaceID, projectID 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
|
||||
@@ -361,6 +371,7 @@ type DashboardService struct {
|
||||
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
|
||||
projectBudget ProjectBudgetEvaluator // optional; nil means reassignment doesn't re-evaluate the destination project's budget policies
|
||||
// officeSessionIdentity gates RecordAgentDecision's use of the caller's
|
||||
// own session id. Defaults false (zero value); set via
|
||||
// SetOfficeSessionIdentity, wired from features.officeSessionIdentity.
|
||||
@@ -480,6 +491,13 @@ func (s *DashboardService) SetRunResolver(r RunResolver) {
|
||||
s.runResolver = r
|
||||
}
|
||||
|
||||
// SetProjectBudgetEvaluator wires the seam used to re-evaluate a
|
||||
// destination project's budget policies on task reassignment. Optional;
|
||||
// when unset, UpdateTaskProjectID does not check budgets.
|
||||
func (s *DashboardService) SetProjectBudgetEvaluator(e ProjectBudgetEvaluator) {
|
||||
s.projectBudget = e
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
||||
@@ -10,6 +10,11 @@ import (
|
||||
"github.com/kandev/kandev/internal/office/repository/sqlite"
|
||||
)
|
||||
|
||||
const (
|
||||
budgetAlertAction = "budget.alert"
|
||||
budgetExceededAction = "budget.exceeded"
|
||||
)
|
||||
|
||||
// GetInboxItems returns a computed view of all items needing user attention.
|
||||
func (s *DashboardService) GetInboxItems(ctx context.Context, wsID string) ([]*models.InboxItem, error) {
|
||||
var items []*models.InboxItem
|
||||
@@ -93,7 +98,7 @@ func (s *DashboardService) GetInboxCount(ctx context.Context, wsID string) (int,
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
alerts, err := s.repo.ListActivityEntriesByAction(ctx, wsID, "budget.alert", 50)
|
||||
alerts, err := s.listBudgetInboxEntries(ctx, wsID, 50)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
@@ -137,16 +142,20 @@ func (s *DashboardService) inboxApprovalItems(ctx context.Context, wsID string)
|
||||
}
|
||||
|
||||
func (s *DashboardService) inboxBudgetAlertItems(ctx context.Context, wsID string) ([]*models.InboxItem, error) {
|
||||
entries, err := s.repo.ListActivityEntriesByAction(ctx, wsID, "budget.alert", 20)
|
||||
entries, err := s.listBudgetInboxEntries(ctx, wsID, 20)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items := make([]*models.InboxItem, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
title := "Budget alert"
|
||||
if e.Action == budgetExceededAction {
|
||||
title = "Budget exceeded"
|
||||
}
|
||||
item := &models.InboxItem{
|
||||
ID: e.ID,
|
||||
Type: "budget_alert",
|
||||
Title: "Budget alert",
|
||||
Title: title,
|
||||
Description: e.Details,
|
||||
Status: "active",
|
||||
EntityID: e.TargetID,
|
||||
@@ -158,6 +167,20 @@ func (s *DashboardService) inboxBudgetAlertItems(ctx context.Context, wsID strin
|
||||
return items, nil
|
||||
}
|
||||
|
||||
func (s *DashboardService) listBudgetInboxEntries(
|
||||
ctx context.Context, wsID string, limit int,
|
||||
) ([]*models.ActivityEntry, error) {
|
||||
alerts, err := s.repo.ListActivityEntriesByAction(ctx, wsID, budgetAlertAction, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
exceeded, err := s.repo.ListActivityEntriesByAction(ctx, wsID, budgetExceededAction, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return append(alerts, exceeded...), nil
|
||||
}
|
||||
|
||||
func (s *DashboardService) inboxAgentErrorItems(ctx context.Context, wsID string) ([]*models.InboxItem, error) {
|
||||
entries, err := s.repo.ListActivityEntriesByAction(ctx, wsID, "agent.error", 20)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
package dashboard_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/events"
|
||||
"github.com/kandev/kandev/internal/events/bus"
|
||||
)
|
||||
|
||||
// projectBudgetEvaluatorSpy implements dashboard.ProjectBudgetEvaluator and
|
||||
// records every call, so UpdateTaskProjectID tests can assert whether (and
|
||||
// with what arguments) the reassignment budget hook fired.
|
||||
type projectBudgetEvaluatorSpy struct {
|
||||
calls []projectBudgetEvaluatorCall
|
||||
err error
|
||||
onEvaluate func()
|
||||
}
|
||||
|
||||
type projectBudgetEvaluatorCall struct {
|
||||
workspaceID string
|
||||
projectID string
|
||||
}
|
||||
|
||||
func (s *projectBudgetEvaluatorSpy) EvaluateProjectBudget(_ context.Context, workspaceID, projectID string) error {
|
||||
s.calls = append(s.calls, projectBudgetEvaluatorCall{workspaceID: workspaceID, projectID: projectID})
|
||||
if s.onEvaluate != nil {
|
||||
s.onEvaluate()
|
||||
}
|
||||
return s.err
|
||||
}
|
||||
|
||||
// projectAssignmentOrderingEventBus records whether the reassignment budget
|
||||
// hook ran before the task-updated event was published.
|
||||
type projectAssignmentOrderingEventBus struct {
|
||||
evaluated *bool
|
||||
published bool
|
||||
evaluatedAtPublish bool
|
||||
}
|
||||
|
||||
func (b *projectAssignmentOrderingEventBus) Publish(_ context.Context, subject string, _ *bus.Event) error {
|
||||
if subject == events.OfficeTaskUpdated {
|
||||
b.published = true
|
||||
b.evaluatedAtPublish = *b.evaluated
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *projectAssignmentOrderingEventBus) Subscribe(string, bus.EventHandler) (bus.Subscription, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (b *projectAssignmentOrderingEventBus) QueueSubscribe(string, string, bus.EventHandler) (bus.Subscription, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (b *projectAssignmentOrderingEventBus) Request(context.Context, string, *bus.Event, time.Duration) (*bus.Event, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (b *projectAssignmentOrderingEventBus) Close() {}
|
||||
func (b *projectAssignmentOrderingEventBus) IsConnected() bool { return true }
|
||||
|
||||
func insertTestProject(t *testing.T, deps *testDeps, id, wsID string) {
|
||||
t.Helper()
|
||||
_, err := deps.db.Exec(`
|
||||
INSERT INTO office_projects (id, workspace_id, name, created_at, updated_at)
|
||||
VALUES (?, ?, ?, datetime('now'), datetime('now'))
|
||||
`, id, wsID, "Test Project "+id)
|
||||
if err != nil {
|
||||
t.Fatalf("insert project %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateTaskProjectID_EvaluatesDestinationProjectBudget covers the
|
||||
// reassignment defect: assigning a task into a project must re-check that
|
||||
// project's budget policies, since #2903 made project-cost rollups follow
|
||||
// the task's live project_id — a reassignment alone can cross a threshold
|
||||
// with no cost event or agent launch to trigger the existing budget hooks.
|
||||
func TestUpdateTaskProjectID_EvaluatesDestinationProjectBudget(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-1", "ws-1", "Task", "todo", 1)
|
||||
insertTestProject(t, deps, "proj-1", "ws-1")
|
||||
|
||||
spy := &projectBudgetEvaluatorSpy{}
|
||||
deps.svc.SetProjectBudgetEvaluator(spy)
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", "proj-1"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID: %v", err)
|
||||
}
|
||||
|
||||
if len(spy.calls) != 1 {
|
||||
t.Fatalf("evaluator calls = %d, want 1", len(spy.calls))
|
||||
}
|
||||
if spy.calls[0].workspaceID != "ws-1" || spy.calls[0].projectID != "proj-1" {
|
||||
t.Errorf("evaluator called with %+v, want {ws-1 proj-1}", spy.calls[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateTaskProjectID_ClearingProjectSkipsEvaluation covers clearing a
|
||||
// project (projectID==""): there is no destination project to evaluate.
|
||||
func TestUpdateTaskProjectID_ClearingProjectSkipsEvaluation(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-1", "ws-1", "Task", "todo", 1)
|
||||
|
||||
spy := &projectBudgetEvaluatorSpy{}
|
||||
deps.svc.SetProjectBudgetEvaluator(spy)
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", ""); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID: %v", err)
|
||||
}
|
||||
|
||||
if len(spy.calls) != 0 {
|
||||
t.Errorf("expected no evaluator calls when clearing project, got %+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateTaskProjectID_UnchangedProjectSkipsEvaluation covers a no-op
|
||||
// reassignment (PATCH project_id to the project the task is already in):
|
||||
// nothing crossed a threshold, so the budget evaluator must not fire. Without
|
||||
// this guard every retried/duplicate PATCH would write a fresh budget.alert
|
||||
// activity row for an already over-threshold project.
|
||||
func TestUpdateTaskProjectID_UnchangedProjectSkipsEvaluation(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-1", "ws-1", "Task", "todo", 1)
|
||||
insertTestProject(t, deps, "proj-1", "ws-1")
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", "proj-1"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID (initial assign): %v", err)
|
||||
}
|
||||
|
||||
spy := &projectBudgetEvaluatorSpy{}
|
||||
deps.svc.SetProjectBudgetEvaluator(spy)
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", "proj-1"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID (no-op reassign): %v", err)
|
||||
}
|
||||
|
||||
if len(spy.calls) != 0 {
|
||||
t.Errorf("expected no evaluator calls for a no-op reassignment, got %+v", spy.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateTaskProjectID_NoEvaluatorWired covers the default (unwired)
|
||||
// dependency: reassignment must still succeed when no evaluator is set.
|
||||
func TestUpdateTaskProjectID_NoEvaluatorWired(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-1", "ws-1", "Task", "todo", 1)
|
||||
insertTestProject(t, deps, "proj-1", "ws-1")
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", "proj-1"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateTaskProjectID_EvaluatorErrorDoesNotFailReassignment covers the
|
||||
// best-effort contract: the tasks.project_id write has already committed by
|
||||
// the time the evaluator runs, so an evaluator error must be swallowed
|
||||
// (logged), not surfaced as a failed reassignment.
|
||||
func TestUpdateTaskProjectID_EvaluatorErrorDoesNotFailReassignment(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-1", "ws-1", "Task", "todo", 1)
|
||||
insertTestProject(t, deps, "proj-1", "ws-1")
|
||||
|
||||
spy := &projectBudgetEvaluatorSpy{err: errors.New("boom")}
|
||||
deps.svc.SetProjectBudgetEvaluator(spy)
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-1", "proj-1"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID should not fail on evaluator error, got: %v", err)
|
||||
}
|
||||
|
||||
task, err := deps.repo.GetTaskByID(context.Background(), "task-1")
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskByID: %v", err)
|
||||
}
|
||||
if task.ProjectID != "proj-1" {
|
||||
t.Errorf("project_id = %q, want proj-1 (write must persist despite evaluator error)", task.ProjectID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateTaskProjectID_EvaluatesBeforePublishingTaskUpdated(t *testing.T) {
|
||||
deps := newTestDeps(t)
|
||||
insertTestTask(t, deps.db, "task-order", "ws-1", "Task", "todo", 1)
|
||||
insertTestProject(t, deps, "proj-order", "ws-1")
|
||||
|
||||
evaluated := false
|
||||
spy := &projectBudgetEvaluatorSpy{onEvaluate: func() { evaluated = true }}
|
||||
eventsBus := &projectAssignmentOrderingEventBus{evaluated: &evaluated}
|
||||
deps.svc.SetProjectBudgetEvaluator(spy)
|
||||
deps.svc.SetEventBus(eventsBus)
|
||||
|
||||
if err := deps.svc.UpdateTaskProjectID(context.Background(), "task-order", "proj-order"); err != nil {
|
||||
t.Fatalf("UpdateTaskProjectID: %v", err)
|
||||
}
|
||||
|
||||
if !eventsBus.published {
|
||||
t.Fatal("expected office.task.updated to be published")
|
||||
}
|
||||
if !eventsBus.evaluatedAtPublish {
|
||||
t.Fatal("budget evaluation must complete before office.task.updated is published")
|
||||
}
|
||||
}
|
||||
@@ -44,8 +44,10 @@ func (s *DashboardService) UpdateTaskPriority(ctx context.Context, taskID, prior
|
||||
// project. When non-empty, validates that the project belongs to the same
|
||||
// workspace as the task.
|
||||
func (s *DashboardService) UpdateTaskProjectID(ctx context.Context, taskID, projectID string) error {
|
||||
var taskWS string
|
||||
if projectID != "" {
|
||||
taskWS, err := s.repo.GetTaskWorkspaceID(ctx, taskID)
|
||||
var err error
|
||||
taskWS, err = s.repo.GetTaskWorkspaceID(ctx, taskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("resolve task workspace: %w", err)
|
||||
}
|
||||
@@ -57,9 +59,35 @@ func (s *DashboardService) UpdateTaskProjectID(ctx context.Context, taskID, proj
|
||||
return fmt.Errorf("project %s belongs to a different workspace", projectID)
|
||||
}
|
||||
}
|
||||
// Read the pre-write project so a no-op PATCH (destination == current)
|
||||
// doesn't trigger a budget re-evaluation below: nothing crossed a
|
||||
// threshold, so evaluatePolicy's unconditional alert/exceeded logging
|
||||
// would otherwise write a fresh activity row on every retry. A read
|
||||
// error is treated as "changed" (fail open to evaluating, the prior
|
||||
// behavior) rather than silently skipping. Only read when a
|
||||
// reassignment could actually trigger an evaluation.
|
||||
needsChangeCheck := projectID != "" && s.projectBudget != nil
|
||||
var prevProjectID string
|
||||
var prevErr error
|
||||
if needsChangeCheck {
|
||||
prevProjectID, prevErr = s.repo.GetTaskProjectID(ctx, taskID)
|
||||
}
|
||||
|
||||
if err := s.repo.UpdateTaskProjectID(ctx, taskID, projectID); err != nil {
|
||||
return err
|
||||
}
|
||||
// Best-effort: the write above has already committed, so an evaluation
|
||||
// error here is logged, not returned. Mirrors event_subscribers.go's
|
||||
// post-cost-event budget check. Evaluate before publishing the task event
|
||||
// so a client's refetch can observe any activity row created by the check.
|
||||
if needsChangeCheck && (prevErr != nil || prevProjectID != projectID) {
|
||||
if err := s.projectBudget.EvaluateProjectBudget(ctx, taskWS, projectID); err != nil {
|
||||
s.logger.Warn("project budget evaluation failed on reassignment",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("project_id", projectID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
s.publishTaskUpdated(ctx, taskID, []string{"project_id"})
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -390,6 +390,21 @@ describe("office WS handler — costs refetch gating on office.task.updated", ()
|
||||
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("tasks");
|
||||
});
|
||||
|
||||
it("refreshes inbox and activity after a task project change", () => {
|
||||
const { store, setOfficeRefetchTrigger } = makeStore(ACTIVE_WS);
|
||||
const handlers = registerOfficeHandlers(store);
|
||||
const handler = handlers[ACTION_TASK_UPDATED]!;
|
||||
|
||||
handler({
|
||||
type: "notification",
|
||||
action: ACTION_TASK_UPDATED,
|
||||
payload: { workspace_id: ACTIVE_WS, task_id: "t-1", fields: ["project_id"] },
|
||||
} as Parameters<typeof handler>[0]);
|
||||
|
||||
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("inbox");
|
||||
expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("activity");
|
||||
});
|
||||
|
||||
it("maps a project_id update to the task store field", () => {
|
||||
const { store, patchTaskInStore } = makeStore(ACTIVE_WS);
|
||||
const handlers = registerOfficeHandlers(store);
|
||||
|
||||
@@ -77,6 +77,8 @@ function buildTaskHandlers(
|
||||
if (Array.isArray(fields) && fields.includes("project_id")) {
|
||||
triggerRefetch("costs");
|
||||
triggerRefetch("tasks");
|
||||
triggerRefetch("inbox");
|
||||
triggerRefetch("activity");
|
||||
}
|
||||
},
|
||||
|
||||
|
||||
Reference in New Issue
Block a user