diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 5deaf7cccf..2cddebfc08 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -820,6 +820,7 @@ func (s *Service) handlePromptUsage(ctx context.Context, event *bus.Event) error return err } s.recordCostEventWritten(string(resolution.source), provider) + s.publishCostRecorded(ctx, fields.WorkspaceID, costEvent) if fields.WorkspaceID != "" { if err := s.CheckBudget( @@ -832,6 +833,30 @@ func (s *Service) handlePromptUsage(ctx context.Context, event *bus.Event) error return nil } +// publishCostRecorded emits the durable cost write as an Office notification. +// The database insert is authoritative; a publish failure is logged and does +// not turn a successful cost write into a failed prompt-usage event. +func (s *Service) publishCostRecorded(ctx context.Context, workspaceID string, costEvent *models.CostEvent) { + if s.eb == nil || workspaceID == "" || costEvent == nil { + return + } + data := map[string]interface{}{ + "workspace_id": workspaceID, + "task_id": costEvent.TaskID, + "session_id": costEvent.SessionID, + "agent_profile_id": costEvent.AgentProfileID, + "project_id": costEvent.ProjectID, + "model": costEvent.Model, + "provider": costEvent.Provider, + "cost_subcents": costEvent.CostSubcents, + } + event := bus.NewEvent(events.OfficeCostRecorded, "office-service", data) + if err := s.eb.Publish(ctx, events.OfficeCostRecorded, event); err != nil { + s.logger.Debug("publish cost recorded event failed", + zap.String("task_id", costEvent.TaskID), zap.Error(err)) + } +} + // resolveProvider derives the provider id for the cost row. AgentType // (the CLI engine slug — claude-acp, codex-acp, ...) is the most reliable // source; AgentID is checked as a fallback for legacy bus events that diff --git a/apps/backend/internal/office/service/event_subscribers_test.go b/apps/backend/internal/office/service/event_subscribers_test.go index 5c9d1fb06b..da8ff8d52b 100644 --- a/apps/backend/internal/office/service/event_subscribers_test.go +++ b/apps/backend/internal/office/service/event_subscribers_test.go @@ -534,6 +534,62 @@ func TestPromptUsage_RecordsCostEvent(t *testing.T) { } } +func TestPromptUsage_PublishesCostRecorded(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + createTestAgent(t, svc, "ws-1", "worker-cost-notification") + insertTestTask(t, svc, "task-cost-notification", "ws-1") + setTestTaskAssignee(t, svc, "task-cost-notification", "worker-cost-notification") + + var recorded *bus.Event + if _, err := eb.Subscribe(events.OfficeCostRecorded, func(_ context.Context, event *bus.Event) error { + recorded = event + return nil + }); err != nil { + t.Fatalf("subscribe cost event: %v", err) + } + + event := bus.NewEvent(events.SessionPromptUsageUpdated, "test", map[string]interface{}{ + "task_id": "task-cost-notification", + "session_id": "session-cost-notification", + "agent_id": "claude-acp", + "model": "gpt-4o-mini", + "usage": map[string]interface{}{ + "input_tokens": 10, + "output_tokens": 5, + }, + }) + if err := eb.Publish(ctx, events.BuildSessionPromptUsageSubject("session-cost-notification"), event); err != nil { + t.Fatalf("publish prompt usage: %v", err) + } + + if recorded == nil { + t.Fatal("expected office.cost.recorded event after cost insert") + } + if recorded.Type != events.OfficeCostRecorded || recorded.Subject != events.OfficeCostRecorded { + t.Fatalf("recorded event identity = type %q subject %q, want %q", recorded.Type, recorded.Subject, events.OfficeCostRecorded) + } + data, ok := recorded.Data.(map[string]interface{}) + if !ok { + t.Fatalf("recorded event data type = %T, want map[string]interface{}", recorded.Data) + } + want := map[string]string{ + "workspace_id": "ws-1", + "task_id": "task-cost-notification", + "session_id": "session-cost-notification", + "project_id": "", + } + for key, value := range want { + if got, _ := data[key].(string); got != value { + t.Errorf("event data[%q] = %q, want %q", key, got, value) + } + } + if got, ok := data["cost_subcents"].(int64); !ok || got != 0 { + t.Errorf("event data[cost_subcents] = %#v, want int64(0)", data["cost_subcents"]) + } +} + // TestPromptUsage_AgentTypeDerivesProvider confirms the production path: // orchestrator's publishPromptUsage populates AgentType from the session // snapshot (claude-acp / codex-acp / ...) while AgentID carries the diff --git a/apps/web/app/office/workspace/costs/cost-overview.test.tsx b/apps/web/app/office/workspace/costs/cost-overview.test.tsx new file mode 100644 index 0000000000..906522977b --- /dev/null +++ b/apps/web/app/office/workspace/costs/cost-overview.test.tsx @@ -0,0 +1,173 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { act, cleanup, render, screen, waitFor } from "@testing-library/react"; +import type { StoreApi } from "zustand"; +import { StateProvider, useAppStoreApi } from "@/components/state-provider"; +import type { AppState } from "@/lib/state/store"; +import { CostOverview } from "./cost-overview"; + +// Hoisted mock so the getCostsBreakdown import is replaced before the +// component imports it. Tests configure the mock per-case. +const getCostsBreakdownMock = vi.hoisted(() => vi.fn()); + +vi.mock("@/lib/api/domains/office-api", async () => { + const actual = await vi.importActual( + "@/lib/api/domains/office-api", + ); + return { + ...actual, + getCostsBreakdown: getCostsBreakdownMock, + }; +}); + +afterEach(() => { + cleanup(); + vi.clearAllMocks(); +}); + +const EMPTY_BREAKDOWN = { + total_subcents: 0, + by_agent: [], + by_project: [], + by_model: [], + by_provider: [], +}; + +// Renders as a StateProvider child purely to hand the real store instance +// back to the test, so it can bump office.refetchTriggers the same way the +// WS handler does and let real zustand reactivity re-render CostOverview. +function StoreCapture({ storeRef }: { storeRef: { current: StoreApi | null } }) { + storeRef.current = useAppStoreApi(); + return null; +} + +describe("CostOverview", () => { + it("refetches the cost breakdown when the costs refetch trigger fires", async () => { + getCostsBreakdownMock.mockResolvedValue(EMPTY_BREAKDOWN); + const storeRef: { current: StoreApi | null } = { current: null }; + + render( + + + + , + ); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(1); + }); + + // Simulates the WS handler bumping "costs" (office.cost.recorded, or + // office.task.updated with fields: ["project_id"]). + act(() => { + storeRef.current!.getState().setOfficeRefetchTrigger("costs"); + }); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(2); + }); + }); + + it("does not refetch on an unrelated refetch trigger", async () => { + getCostsBreakdownMock.mockResolvedValue(EMPTY_BREAKDOWN); + const storeRef: { current: StoreApi | null } = { current: null }; + + render( + + + + , + ); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(1); + }); + + act(() => { + storeRef.current!.getState().setOfficeRefetchTrigger("dashboard"); + }); + + // Give any (incorrect) refetch a chance to fire before asserting it didn't. + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(1); + }); + + it("discards a stale response when an older request resolves after a newer one", async () => { + let resolveFirst: (value: typeof EMPTY_BREAKDOWN) => void; + let resolveSecond: (value: typeof EMPTY_BREAKDOWN) => void; + getCostsBreakdownMock + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirst = resolve; + }), + ) + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveSecond = resolve; + }), + ); + const storeRef: { current: StoreApi | null } = { current: null }; + + render( + + + + , + ); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(1); + }); + + act(() => { + storeRef.current!.getState().setOfficeRefetchTrigger("costs"); + }); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(2); + }); + + // Resolve the newer (second) request first with real data, then the + // stale first request with different data. The stale response must not + // overwrite the fresher state. + await act(async () => { + resolveSecond({ ...EMPTY_BREAKDOWN, total_subcents: 20000 }); + }); + await act(async () => { + resolveFirst({ ...EMPTY_BREAKDOWN, total_subcents: 10000 }); + }); + + // getByText throws if no element with this text exists. + screen.getByText("$2.00"); + }); +}); + +describe("CostOverview workspace isolation", () => { + it("does not show the previous workspace after the next workspace fails to load", async () => { + getCostsBreakdownMock + .mockResolvedValueOnce({ ...EMPTY_BREAKDOWN, total_subcents: 10000 }) + .mockRejectedValueOnce(new Error("workspace unavailable")); + + const view = render( + + + , + ); + + await waitFor(() => { + expect(screen.getByText("$1.00")).toBeTruthy(); + }); + + view.rerender( + + + , + ); + + await waitFor(() => { + expect(getCostsBreakdownMock).toHaveBeenCalledTimes(2); + }); + expect(screen.queryByText("$1.00")).toBeNull(); + }); +}); diff --git a/apps/web/app/office/workspace/costs/cost-overview.tsx b/apps/web/app/office/workspace/costs/cost-overview.tsx index 35866bfe99..22b9936de6 100644 --- a/apps/web/app/office/workspace/costs/cost-overview.tsx +++ b/apps/web/app/office/workspace/costs/cost-overview.tsx @@ -1,6 +1,6 @@ "use client"; -import { useEffect, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { Button } from "@kandev/ui/button"; import { IconCurrencyDollar, @@ -10,6 +10,7 @@ import { IconBuilding, } from "@tabler/icons-react"; import { toast } from "@/lib/toast/sonner"; +import { useOfficeRefetch } from "@/hooks/use-office-refetch"; import { getCostsBreakdown } from "@/lib/api/domains/office-api"; import { MetricCard } from "../../components/metric-card"; import type { CostBreakdownItem } from "@/lib/state/slices/office/types"; @@ -23,20 +24,43 @@ import { t as staticT } from "@/lib/i18n"; type DateRange = "mtd" | "30d"; +type CostOverviewData = { + totalSubcents: number; + byAgent: CostBreakdownItem[]; + byProject: CostBreakdownItem[]; + byModel: CostBreakdownItem[]; + byProvider: CostBreakdownItem[]; +}; + +const EMPTY_COST_OVERVIEW: CostOverviewData = { + totalSubcents: 0, + byAgent: [], + byProject: [], + byModel: [], + byProvider: [], +}; + export function CostOverview({ workspaceId }: { workspaceId: string }) { - const { t } = useTranslation(); const [range, setRange] = useState("mtd"); const [totalSubcents, setTotalSubcents] = useState(0); const [byAgent, setByAgent] = useState([]); const [byProject, setByProject] = useState([]); const [byModel, setByModel] = useState([]); const [byProvider, setByProvider] = useState([]); + const [loadedWorkspaceId, setLoadedWorkspaceId] = useState(null); - useEffect(() => { + // Range changes, workspace changes, and WS-triggered refetches (below) can + // all call fetchCosts in overlapping succession; guard against an older + // request's response landing after a newer one and clobbering fresher data. + const requestId = useRef(0); + const fetchCosts = useCallback(() => { + const current = ++requestId.current; // Single composed call (Stream D of office optimization). Was four // parallel round-trips (summary + by-agent + by-project + by-model). getCostsBreakdown(workspaceId) .then((res) => { + if (current !== requestId.current) return; + setLoadedWorkspaceId(workspaceId); setTotalSubcents(res.total_subcents ?? 0); setByAgent((res.by_agent ?? []) as CostBreakdownItem[]); setByProject((res.by_project ?? []) as CostBreakdownItem[]); @@ -44,10 +68,47 @@ export function CostOverview({ workspaceId }: { workspaceId: string }) { setByProvider((res.by_provider ?? []) as CostBreakdownItem[]); }) .catch((err) => { + if (current !== requestId.current) return; toast.error(err instanceof Error ? err.message : staticT("office:failedToLoadCostData")); }); - }, [workspaceId, range]); + }, [workspaceId]); + useEffect(() => { + fetchCosts(); + }, [fetchCosts, range]); + + // The WS handler bumps "costs" on a new cost event (office.cost.recorded) + // and on a task project reassignment (office.task.updated with + // fields: ["project_id"]) — both change the by-project breakdown for an + // already-open tab. + useOfficeRefetch("costs", fetchCosts); + + // Do not render a previous workspace's response while a new workspace is + // loading. This also keeps the old values hidden when the new request + // fails, instead of showing them under the wrong workspace. + const displayData = + loadedWorkspaceId === workspaceId + ? { totalSubcents, byAgent, byProject, byModel, byProvider } + : EMPTY_COST_OVERVIEW; + + return ; +} + +type CostOverviewContentProps = CostOverviewData & { + range: DateRange; + onRangeChange: (range: DateRange) => void; +}; + +function CostOverviewContent({ + range, + onRangeChange, + totalSubcents, + byAgent, + byProject, + byModel, + byProvider, +}: CostOverviewContentProps) { + const { t } = useTranslation(); return (
@@ -55,7 +116,7 @@ export function CostOverview({ workspaceId }: { workspaceId: string }) { variant={range === "mtd" ? "secondary" : "outline"} size="sm" className="cursor-pointer" - onClick={() => setRange("mtd")} + onClick={() => onRangeChange("mtd")} > MTD @@ -63,7 +124,7 @@ export function CostOverview({ workspaceId }: { workspaceId: string }) { variant={range === "30d" ? "secondary" : "outline"} size="sm" className="cursor-pointer" - onClick={() => setRange("30d")} + onClick={() => onRangeChange("30d")} > {t("office:last30Days")} diff --git a/apps/web/e2e/helpers/office-api-client.ts b/apps/web/e2e/helpers/office-api-client.ts index b9a798cc6f..e06ba8d781 100644 --- a/apps/web/e2e/helpers/office-api-client.ts +++ b/apps/web/e2e/helpers/office-api-client.ts @@ -171,6 +171,30 @@ export class OfficeApiClient { return this.request("GET", `/workspaces/${wsId}/tasks/search?q=${encodeURIComponent(query)}`); } + /** + * Reassign (or clear, with "") the task's project. Publishes + * `office.task.updated` with `fields: ["project_id"]`, which is what the + * costs page's live refetch gate keys off. + */ + async updateTaskProjectId(id: string, projectId: string): Promise> { + return this.request("PATCH", `/tasks/${id}`, { project_id: projectId }); + } + + /** + * Create a project in the workspace. Onboarding does not seed a default + * project (see e2e/tests/office/project-repository-picker.spec.ts), so + * tests that need a real project_id to reassign a task to must create + * one first. + */ + async createProject(wsId: string, name: string): Promise> { + const res = await this.request<{ project?: Record }>( + "POST", + `/workspaces/${wsId}/projects`, + { name }, + ); + return res.project ?? (res as unknown as Record); + } + // --- Labels --- async addLabel(wsId: string, taskId: string, name: string): Promise> { diff --git a/apps/web/e2e/tests/office/costs.spec.ts b/apps/web/e2e/tests/office/costs.spec.ts index 5a7ce4f31b..e4603fcc0e 100644 --- a/apps/web/e2e/tests/office/costs.spec.ts +++ b/apps/web/e2e/tests/office/costs.spec.ts @@ -1,5 +1,6 @@ import { test, expect } from "../../fixtures/office-fixture"; import { officeTopbarTitle } from "../../helpers/office-topbar"; +import { waitForHttp } from "../../helpers/causal-waits"; test.describe("Costs & Budgets", () => { test("cost summary starts at zero", async ({ officeApi, officeSeed }) => { @@ -18,4 +19,41 @@ test.describe("Costs & Budgets", () => { timeout: 10_000, }); }); + + test("costs page refetches the breakdown live when a task's project is reassigned", async ({ + testPage, + officeApi, + officeSeed, + }) => { + const task = (await officeApi.createTask(officeSeed.workspaceId, "Costs Refetch Task", { + workflow_id: officeSeed.workflowId, + })) as Record; + const taskId = task.id as string; + + // Onboarding does not seed a default project, so a real project must be + // created before it can be assigned to the task. + const project = (await officeApi.createProject( + officeSeed.workspaceId, + "Costs Refetch Project", + )) as Record; + const projectId = project.id as string; + + const breakdownPath = /\/costs\/breakdown$/; + const initialLoad = waitForHttp(testPage, "GET", breakdownPath); + await testPage.goto("/office/workspace/costs"); + await expect(officeTopbarTitle(testPage)).toHaveText(/Costs/i, { timeout: 10_000 }); + await initialLoad; + + // Reassigning the task's project publishes office.task.updated with + // fields: ["project_id"] (DashboardService.UpdateTaskProjectID). The + // costs page's WS handler must gate a "costs" refetch trigger on that, + // so an already-open tab sees the new breakdown without a manual + // reload — this is the live-UI staleness gap from PR #2903's follow-up. + // Assert the actual network refetch fires, not just that the WS frame + // arrived, so a regression in the store trigger -> useOfficeRefetch wiring + // fails this test too. + const refetch = waitForHttp(testPage, "GET", breakdownPath); + await officeApi.updateTaskProjectId(taskId, projectId); + await refetch; + }); }); diff --git a/apps/web/lib/ws/handlers/office.test.ts b/apps/web/lib/ws/handlers/office.test.ts index 04f442c152..6bf305b81e 100644 --- a/apps/web/lib/ws/handlers/office.test.ts +++ b/apps/web/lib/ws/handlers/office.test.ts @@ -343,6 +343,72 @@ describe("office WS handler — content-editing guard (AC-61)", () => { }); }); +describe("office WS handler — costs refetch gating on office.task.updated", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + // Cost-by-project breakdown groups by the task's live project_id (see + // costs.go), so only a project_id field change should bump "costs" — the + // generic kanban task.updated forward carries no `fields` at all and must + // not refetch costs on every unrelated task touch. + it.each([ + { fields: ["project_id"], expectCosts: true }, + { fields: ["project_id", "title"], expectCosts: true }, + { fields: ["priority"], expectCosts: false }, + { fields: [], expectCosts: false }, + { fields: undefined, expectCosts: false }, + ])("gates the costs trigger by fields=$fields", ({ fields, expectCosts }) => { + 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 }, + } as Parameters[0]); + + if (expectCosts) { + expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("costs"); + } else { + expect(setOfficeRefetchTrigger).not.toHaveBeenCalledWith("costs"); + } + }); + + it("refreshes tasks when a task project changes", () => { + 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[0]); + + expect(setOfficeRefetchTrigger).toHaveBeenCalledWith("tasks"); + }); + + it("maps a project_id update to the task store field", () => { + const { store, patchTaskInStore } = 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", + project_id: "project-2", + }, + } as Parameters[0]); + + expect(patchTaskInStore).toHaveBeenCalledWith("t-1", { projectId: "project-2" }); + }); +}); + describe("office WS handler — approval flow events", () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/apps/web/lib/ws/handlers/office.ts b/apps/web/lib/ws/handlers/office.ts index d5c07d7b51..a5fa990639 100644 --- a/apps/web/lib/ws/handlers/office.ts +++ b/apps/web/lib/ws/handlers/office.ts @@ -67,6 +67,17 @@ function buildTaskHandlers( // server-authoritative task DTO after a property mutation. triggerRefetch(`task:${taskId}`); triggerRefetch("dashboard"); + // Cost-by-project breakdown groups by the task's live project_id + // (see costs.go), so reassigning a task's project changes an + // already-computed breakdown without a dedicated cost event. Gate on + // `fields` (only DashboardService.UpdateTaskProjectID sends it) so the + // generic kanban task.updated forward — which carries no `fields` — + // doesn't refetch costs on every unrelated task touch. + const fields = p.fields as string[] | undefined; + if (Array.isArray(fields) && fields.includes("project_id")) { + triggerRefetch("costs"); + triggerRefetch("tasks"); + } }, "office.task.created": (message) => { @@ -299,6 +310,7 @@ function normalizeIssueFields(p: Record): Record