From efe83dc93521e2bdce7254f167dd244350719895 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 9 Jul 2026 16:09:35 +0000 Subject: [PATCH] Cherry-pick server attention feed endpoint (PAP-13205/13206) Dependency of the What-needs-me page: adds AttentionFeed/AttentionItem shared types + GET /api/companies/:id/attention. Original commit e35b11332 (now in master). Co-Authored-By: Paperclip --- packages/shared/src/index.ts | 10 + packages/shared/src/types/attention.ts | 68 ++ packages/shared/src/types/index.ts | 9 + .../src/__tests__/attention-service.test.ts | 638 +++++++++++++ server/src/__tests__/openapi-routes.test.ts | 1 + server/src/app.ts | 2 + server/src/routes/attention.ts | 28 + server/src/routes/inbox-dismissals.ts | 2 +- server/src/routes/index.ts | 1 + server/src/routes/openapi.ts | 11 +- server/src/services/attention.ts | 843 ++++++++++++++++++ server/src/services/index.ts | 1 + 12 files changed, 1612 insertions(+), 2 deletions(-) create mode 100644 packages/shared/src/types/attention.ts create mode 100644 server/src/__tests__/attention-service.test.ts create mode 100644 server/src/routes/attention.ts create mode 100644 server/src/services/attention.ts diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index d4ef35a9b4d..fe8b4fecc4c 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -49,6 +49,16 @@ export { type ResponsibleUserDenialCopy, type ResponsibleUserDenialTone, } from "./responsible-user-denial.js"; +export type { + AttentionDecisionVerb, + AttentionFeed, + AttentionItem, + AttentionSeverity, + AttentionSourceKind, + AttentionSubject, + AttentionSubjectKind, +} from "./types/attention.js"; + export type { PipelineAutomationRetryBlocker, PipelineAutomationRetryCleanupOptions, diff --git a/packages/shared/src/types/attention.ts b/packages/shared/src/types/attention.ts new file mode 100644 index 00000000000..60d4bd44413 --- /dev/null +++ b/packages/shared/src/types/attention.ts @@ -0,0 +1,68 @@ +export type AttentionSourceKind = + | "approval" + | "issue_thread_interaction" + | "join_request" + | "recovery_action" + | "productivity_review" + | "blocker_attention" + | "review" + | "failed_run" + | "budget_alert" + | "agent_error_alert"; + +export type AttentionSubjectKind = + | "approval" + | "issue" + | "interaction" + | "join_request" + | "recovery_action" + | "run" + | "budget_incident" + | "agent"; + +export type AttentionSeverity = "critical" | "high" | "medium" | "low"; + +export interface AttentionSubject { + kind: AttentionSubjectKind; + id: string; + companyId: string; + title: string | null; + identifier: string | null; + status: string | null; + href: string | null; + metadata?: Record; +} + +export interface AttentionDecisionVerb { + id: string; + label: string; + description: string | null; +} + +export interface AttentionItem { + id: string; + companyId: string; + sourceKind: AttentionSourceKind; + subject: AttentionSubject; + whyNow: string; + decisionVerbs: AttentionDecisionVerb[]; + inlineResolvable: boolean; + entryRule: string; + exitRule: string; + dedupKey: string; + dismissalKey: string; + severity: AttentionSeverity; + rank: number; + activityAt: string; + createdAt: string; + updatedAt: string; + relatedIssue: AttentionSubject | null; +} + +export interface AttentionFeed { + companyId: string; + generatedAt: string; + totalCount: number; + countsBySourceKind: Record; + items: AttentionItem[]; +} diff --git a/packages/shared/src/types/index.ts b/packages/shared/src/types/index.ts index d312edc0244..14c9abe627f 100644 --- a/packages/shared/src/types/index.ts +++ b/packages/shared/src/types/index.ts @@ -1,4 +1,13 @@ export type { Company } from "./company.js"; +export type { + AttentionDecisionVerb, + AttentionFeed, + AttentionItem, + AttentionSeverity, + AttentionSourceKind, + AttentionSubject, + AttentionSubjectKind, +} from "./attention.js"; export type { Environment, EnvironmentDeleteBlastRadius, diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts new file mode 100644 index 00000000000..c731b3ca56c --- /dev/null +++ b/server/src/__tests__/attention-service.test.ts @@ -0,0 +1,638 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + activityLog, + agents, + approvals, + budgetIncidents, + budgetPolicies, + companies, + createDb, + heartbeatRunEvents, + heartbeatRuns, + inboxDismissals, + invites, + issueApprovals, + issueRecoveryActions, + issueRelations, + issueThreadInteractions, + issues, + joinRequests, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { errorHandler } from "../middleware/index.js"; +import { attentionRoutes } from "../routes/attention.js"; +import { attentionService } from "../services/attention.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres attention service tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("attention service", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-"); + db = createDb(tempDb.connectionString); + }, 30_000); + + afterEach(async () => { + await db.delete(inboxDismissals); + await db.delete(issueThreadInteractions); + await db.delete(issueApprovals); + await db.delete(heartbeatRunEvents); + await db.delete(heartbeatRuns); + await db.delete(budgetIncidents); + await db.delete(budgetPolicies); + await db.delete(joinRequests); + await db.delete(invites); + await db.delete(issueRecoveryActions); + await db.delete(issueRelations); + await db.delete(activityLog); + await db.delete(approvals); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedCompany(prefix = "ATN") { + const companyId = randomUUID(); + const workerId = randomUUID(); + const reviewerId = randomUUID(); + const errorAgentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: `${prefix} Co`, + issuePrefix: prefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values([ + { + id: workerId, + companyId, + name: "Worker", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: reviewerId, + companyId, + name: "Reviewer", + role: "qa", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: errorAgentId, + companyId, + name: "Broken Agent", + role: "engineer", + status: "error", + errorReason: "adapter config missing", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + return { companyId, workerId, reviewerId, errorAgentId, prefix }; + } + + async function insertIssue(input: { + companyId: string; + id?: string; + identifier: string; + title: string; + status: string; + priority?: string; + parentId?: string | null; + assigneeAgentId?: string | null; + assigneeUserId?: string | null; + originKind?: string; + originId?: string | null; + originFingerprint?: string; + executionState?: Record | null; + updatedAt?: Date; + createdAt?: Date; + }) { + const id = input.id ?? randomUUID(); + await db.insert(issues).values({ + id, + companyId: input.companyId, + identifier: input.identifier, + title: input.title, + status: input.status, + priority: input.priority ?? "medium", + parentId: input.parentId ?? null, + assigneeAgentId: input.assigneeAgentId ?? null, + assigneeUserId: input.assigneeUserId ?? null, + originKind: input.originKind ?? "manual", + originId: input.originId ?? null, + originFingerprint: input.originFingerprint ?? "default", + executionState: input.executionState ?? null, + createdAt: input.createdAt, + updatedAt: input.updatedAt, + }); + return id; + } + + function pendingUserExecutionState(userId = "board-user") { + return { + status: "pending", + currentStageId: null, + currentStageIndex: null, + currentStageType: "review", + currentParticipant: { type: "user", userId }, + returnAssignee: null, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + monitor: null, + }; + } + + function pendingAgentExecutionState(agentId: string) { + return { + ...pendingUserExecutionState(), + currentParticipant: { type: "agent", agentId }, + }; + } + + it("returns ranked decision-only items for every active source and excludes non-human or transient rows", async () => { + const { companyId, workerId, reviewerId } = await seedCompany("ATN"); + const baseTime = new Date("2026-07-09T12:00:00.000Z"); + const interactionIssueId = await insertIssue({ + companyId, + identifier: "ATN-1", + title: "Needs interaction", + status: "in_progress", + assigneeAgentId: workerId, + updatedAt: baseTime, + }); + const recoverySourceIssueId = await insertIssue({ + companyId, + identifier: "ATN-2", + title: "Needs recovery", + status: "in_progress", + assigneeAgentId: workerId, + updatedAt: baseTime, + }); + const agentRecoverySourceIssueId = await insertIssue({ + companyId, + identifier: "ATN-21", + title: "Agent-owned recovery source", + status: "in_progress", + assigneeAgentId: workerId, + updatedAt: baseTime, + }); + const productivitySourceIssueId = await insertIssue({ + companyId, + identifier: "ATN-3", + title: "Needs productivity review source", + status: "in_progress", + assigneeAgentId: workerId, + updatedAt: baseTime, + }); + const agentProductivitySourceIssueId = await insertIssue({ + companyId, + identifier: "ATN-31", + title: "Agent productivity review source", + status: "in_progress", + assigneeAgentId: workerId, + updatedAt: baseTime, + }); + const blockerParentId = await insertIssue({ + companyId, + identifier: "ATN-4", + title: "Blocked parent", + status: "blocked", + updatedAt: new Date("2026-07-09T12:04:00.000Z"), + }); + const blockerLeafId = await insertIssue({ + companyId, + identifier: "ATN-5", + title: "Stalled review blocker", + status: "in_review", + assigneeAgentId: reviewerId, + updatedAt: new Date("2026-07-09T12:05:00.000Z"), + }); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerLeafId, + relatedIssueId: blockerParentId, + type: "blocks", + }); + const reviewUserIssueId = await insertIssue({ + companyId, + identifier: "ATN-6", + title: "Human review", + status: "in_review", + executionState: pendingUserExecutionState(), + updatedAt: new Date("2026-07-09T12:06:00.000Z"), + }); + await insertIssue({ + companyId, + identifier: "ATN-7", + title: "Agent review excluded", + status: "in_review", + executionState: pendingAgentExecutionState(reviewerId), + updatedAt: new Date("2026-07-09T12:07:00.000Z"), + }); + + const pendingApprovalId = randomUUID(); + await db.insert(approvals).values([ + { + id: pendingApprovalId, + companyId, + type: "hire_agent", + status: "pending", + payload: { title: "Hire Designer" }, + createdAt: new Date("2026-07-09T12:01:00.000Z"), + updatedAt: new Date("2026-07-09T12:01:00.000Z"), + }, + { + id: randomUUID(), + companyId, + type: "hire_agent", + status: "revision_requested", + payload: { title: "Revision requested" }, + createdAt: new Date("2026-07-09T12:02:00.000Z"), + updatedAt: new Date("2026-07-09T12:02:00.000Z"), + }, + ]); + + await db.insert(issueThreadInteractions).values([ + { + id: randomUUID(), + companyId, + issueId: interactionIssueId, + kind: "ask_user_questions", + status: "pending", + continuationPolicy: "wake_assignee", + title: "Pick a launch date", + payload: { version: 1, questions: [] }, + createdAt: new Date("2026-07-09T12:03:00.000Z"), + updatedAt: new Date("2026-07-09T12:03:00.000Z"), + }, + { + id: randomUUID(), + companyId, + issueId: interactionIssueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee", + title: "Already accepted", + payload: { version: 1, prompt: "Already done" }, + createdAt: new Date("2026-07-09T12:03:30.000Z"), + updatedAt: new Date("2026-07-09T12:03:30.000Z"), + }, + ]); + + const inviteId = randomUUID(); + await db.insert(invites).values({ + id: inviteId, + companyId, + tokenHash: `hash-${inviteId}`, + allowedJoinTypes: "both", + expiresAt: new Date("2026-07-10T00:00:00.000Z"), + }); + await db.insert(joinRequests).values({ + id: randomUUID(), + inviteId, + companyId, + requestType: "human", + status: "pending_approval", + requestIp: "127.0.0.1", + requestEmailSnapshot: "new@paperclip.test", + createdAt: new Date("2026-07-09T12:04:00.000Z"), + updatedAt: new Date("2026-07-09T12:04:00.000Z"), + }); + + await db.insert(issueRecoveryActions).values([ + { + id: randomUUID(), + companyId, + sourceIssueId: recoverySourceIssueId, + kind: "missing_disposition", + status: "escalated", + ownerType: "board", + ownerAgentId: null, + ownerUserId: null, + cause: "missing_disposition", + fingerprint: "human-recovery", + evidence: {}, + nextAction: "Choose the final disposition.", + createdAt: new Date("2026-07-09T12:05:00.000Z"), + updatedAt: new Date("2026-07-09T12:05:00.000Z"), + }, + { + id: randomUUID(), + companyId, + sourceIssueId: agentRecoverySourceIssueId, + kind: "stranded_assigned_issue", + status: "active", + ownerType: "agent", + ownerAgentId: workerId, + ownerUserId: null, + cause: "stranded", + fingerprint: "agent-recovery", + evidence: {}, + nextAction: "Agent should self-heal.", + createdAt: new Date("2026-07-09T12:05:30.000Z"), + updatedAt: new Date("2026-07-09T12:05:30.000Z"), + }, + ]); + + await insertIssue({ + companyId, + identifier: "ATN-8", + title: "Human productivity review", + status: "todo", + priority: "high", + parentId: productivitySourceIssueId, + assigneeUserId: "board-user", + originKind: "issue_productivity_review", + originId: productivitySourceIssueId, + originFingerprint: `productivity-review:${productivitySourceIssueId}`, + updatedAt: new Date("2026-07-09T12:08:00.000Z"), + }); + await insertIssue({ + companyId, + identifier: "ATN-9", + title: "Agent productivity review excluded", + status: "todo", + priority: "high", + parentId: agentProductivitySourceIssueId, + assigneeAgentId: workerId, + originKind: "issue_productivity_review", + originId: agentProductivitySourceIssueId, + originFingerprint: `productivity-review-agent:${agentProductivitySourceIssueId}`, + updatedAt: new Date("2026-07-09T12:08:30.000Z"), + }); + + const exhaustedRunId = randomUUID(); + const transientRunId = randomUUID(); + await db.insert(heartbeatRuns).values([ + { + id: exhaustedRunId, + companyId, + agentId: workerId, + invocationSource: "automation", + status: "failed", + error: "adapter failed", + errorCode: "adapter_failed", + contextSnapshot: { issueId: reviewUserIssueId }, + scheduledRetryAttempt: 4, + scheduledRetryReason: "transient_failure", + createdAt: new Date("2026-07-09T12:09:00.000Z"), + updatedAt: new Date("2026-07-09T12:09:00.000Z"), + finishedAt: new Date("2026-07-09T12:09:00.000Z"), + }, + { + id: transientRunId, + companyId, + agentId: reviewerId, + invocationSource: "automation", + status: "failed", + error: "will retry", + errorCode: "provider_quota", + contextSnapshot: { issueId: interactionIssueId }, + createdAt: new Date("2026-07-09T12:09:30.000Z"), + updatedAt: new Date("2026-07-09T12:09:30.000Z"), + finishedAt: new Date("2026-07-09T12:09:30.000Z"), + }, + ]); + await db.insert(heartbeatRunEvents).values({ + companyId, + runId: exhaustedRunId, + agentId: workerId, + seq: 1, + eventType: "lifecycle", + message: "Bounded retry exhausted after 4 scheduled attempts; no further automatic retry will be queued", + payload: { retryReason: "transient_failure", maxAttempts: 4 }, + createdAt: new Date("2026-07-09T12:09:01.000Z"), + }); + + const softPolicy85Id = randomUUID(); + const softPolicy84Id = randomUUID(); + const hardPolicyId = randomUUID(); + await db.insert(budgetPolicies).values([ + { + id: softPolicy85Id, + companyId, + scopeType: "company", + scopeId: companyId, + metric: "billed_cents", + windowKind: "calendar_month_utc", + amount: 100, + }, + { + id: softPolicy84Id, + companyId, + scopeType: "company", + scopeId: companyId, + metric: "billed_cents", + windowKind: "lifetime", + amount: 100, + }, + { + id: hardPolicyId, + companyId, + scopeType: "agent", + scopeId: workerId, + metric: "billed_cents", + windowKind: "calendar_month_utc", + amount: 100, + }, + ]); + await db.insert(budgetIncidents).values([ + { + companyId, + policyId: softPolicy85Id, + scopeType: "company", + scopeId: companyId, + metric: "billed_cents", + windowKind: "calendar_month_utc", + windowStart: new Date("2026-07-01T00:00:00.000Z"), + windowEnd: new Date("2026-08-01T00:00:00.000Z"), + thresholdType: "soft", + amountLimit: 100, + amountObserved: 85, + status: "open", + createdAt: new Date("2026-07-09T12:10:00.000Z"), + updatedAt: new Date("2026-07-09T12:10:00.000Z"), + }, + { + companyId, + policyId: softPolicy84Id, + scopeType: "company", + scopeId: companyId, + metric: "billed_cents", + windowKind: "lifetime", + windowStart: new Date("1970-01-01T00:00:00.000Z"), + windowEnd: new Date("9999-01-01T00:00:00.000Z"), + thresholdType: "soft", + amountLimit: 100, + amountObserved: 84, + status: "open", + createdAt: new Date("2026-07-09T12:10:30.000Z"), + updatedAt: new Date("2026-07-09T12:10:30.000Z"), + }, + { + companyId, + policyId: hardPolicyId, + scopeType: "agent", + scopeId: workerId, + metric: "billed_cents", + windowKind: "calendar_month_utc", + windowStart: new Date("2026-07-01T00:00:00.000Z"), + windowEnd: new Date("2026-08-01T00:00:00.000Z"), + thresholdType: "hard", + amountLimit: 100, + amountObserved: 100, + status: "open", + createdAt: new Date("2026-07-09T12:11:00.000Z"), + updatedAt: new Date("2026-07-09T12:11:00.000Z"), + }, + ]); + + const feed = await attentionService(db).list(companyId, { userId: "board-user" }); + + expect(feed.totalCount).toBe(11); + expect(feed.countsBySourceKind).toMatchObject({ + approval: 1, + issue_thread_interaction: 1, + join_request: 1, + recovery_action: 1, + productivity_review: 1, + blocker_attention: 1, + review: 1, + failed_run: 1, + budget_alert: 2, + agent_error_alert: 1, + }); + expect(feed.items.map((item) => item.sourceKind)).toEqual(expect.arrayContaining([ + "approval", + "issue_thread_interaction", + "join_request", + "recovery_action", + "productivity_review", + "blocker_attention", + "review", + "failed_run", + "budget_alert", + "agent_error_alert", + ])); + for (const item of feed.items) { + expect(item.dedupKey).toBeTruthy(); + expect(item.dismissalKey).toBe(`attention:${item.dedupKey}`); + expect(item.whyNow).toBeTruthy(); + expect(item.entryRule).toBeTruthy(); + expect(item.exitRule).toBeTruthy(); + expect(item.decisionVerbs.length).toBeGreaterThan(0); + expect(item.rank).toBeGreaterThan(0); + } + expect(feed.items.some((item) => item.subject.title === "Revision requested")).toBe(false); + expect(feed.items.some((item) => item.subject.title === "Agent productivity review excluded")).toBe(false); + expect(feed.items.some((item) => item.subject.title === "Agent review excluded")).toBe(false); + expect(feed.items.some((item) => + item.sourceKind === "failed_run" && item.subject.metadata?.errorCode === "provider_quota" + )).toBe(false); + expect(feed.items.some((item) => + item.sourceKind === "budget_alert" && item.subject.metadata?.observedPercent === 84 + )).toBe(false); + }); + + it("uses inbox_dismissals with attention-prefixed dedup keys and resurfaces newer activity", async () => { + const { companyId } = await seedCompany("ATD"); + const approvalId = randomUUID(); + await db.insert(approvals).values({ + id: approvalId, + companyId, + type: "hire_agent", + status: "pending", + payload: { title: "Hire Writer" }, + createdAt: new Date("2026-07-09T12:00:00.000Z"), + updatedAt: new Date("2026-07-09T12:00:00.000Z"), + }); + await db.insert(inboxDismissals).values({ + companyId, + userId: "board-user", + itemKey: `attention:approval:${approvalId}`, + dismissedAt: new Date("2026-07-09T13:00:00.000Z"), + }); + + await expect(attentionService(db).list(companyId, { userId: "board-user" })) + .resolves.toMatchObject({ totalCount: 1 }); // agent_error_alert from seed + await expect(attentionService(db).list(companyId, { userId: "board-user", includeDismissed: true })) + .resolves.toMatchObject({ totalCount: 2 }); + + await db + .update(approvals) + .set({ updatedAt: new Date("2026-07-09T14:00:00.000Z") }) + .where(eq(approvals.id, approvalId)); + + const feed = await attentionService(db).list(companyId, { userId: "board-user" }); + expect(feed.items.some((item) => item.dedupKey === `approval:${approvalId}`)).toBe(true); + }); + + it("serves the route for board users and rejects agent callers", async () => { + const { companyId } = await seedCompany("ATR"); + + function app(actor: Record) { + const testApp = express(); + testApp.use(express.json()); + testApp.use((req, _res, next) => { + (req as any).actor = actor; + next(); + }); + testApp.use("/api", attentionRoutes(db)); + testApp.use(errorHandler); + return testApp; + } + + const board = { + type: "board", + source: "local_implicit", + userId: "board-user", + companyIds: [companyId], + isInstanceAdmin: false, + }; + const agent = { + type: "agent", + source: "agent_key", + companyId, + agentId: randomUUID(), + runId: null, + }; + + await request(app(board)).get(`/api/companies/${companyId}/attention`).expect(200); + await request(app(agent)).get(`/api/companies/${companyId}/attention`).expect(403); + }); +}); diff --git a/server/src/__tests__/openapi-routes.test.ts b/server/src/__tests__/openapi-routes.test.ts index 5eb8c8e87e7..d7433023072 100644 --- a/server/src/__tests__/openapi-routes.test.ts +++ b/server/src/__tests__/openapi-routes.test.ts @@ -15,6 +15,7 @@ const apiPrefixes: Record = { "activity.ts": "/api", "adapters.ts": "/api", "agents.ts": "/api", + "attention.ts": "/api", "approvals.ts": "/api", "assets.ts": "/api", "auth.ts": "/api/auth", diff --git a/server/src/app.ts b/server/src/app.ts index 75e0f1fe243..9cb4ad9f9da 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -33,6 +33,7 @@ import { secretRoutes } from "./routes/secrets.js"; import { costRoutes } from "./routes/costs.js"; import { activityRoutes } from "./routes/activity.js"; import { dashboardRoutes } from "./routes/dashboard.js"; +import { attentionRoutes } from "./routes/attention.js"; import { userProfileRoutes } from "./routes/user-profiles.js"; import { sidebarBadgeRoutes } from "./routes/sidebar-badges.js"; import { sidebarPreferenceRoutes } from "./routes/sidebar-preferences.js"; @@ -253,6 +254,7 @@ export async function createApp( api.use(costRoutes(db, { pluginWorkerManager: workerManager })); api.use(activityRoutes(db)); api.use(dashboardRoutes(db)); + api.use(attentionRoutes(db)); api.use(userProfileRoutes(db)); api.use(sidebarBadgeRoutes(db)); api.use(sidebarPreferenceRoutes(db)); diff --git a/server/src/routes/attention.ts b/server/src/routes/attention.ts new file mode 100644 index 00000000000..a8ef20f0e09 --- /dev/null +++ b/server/src/routes/attention.ts @@ -0,0 +1,28 @@ +import { Router } from "express"; +import type { Db } from "@paperclipai/db"; +import { attentionService } from "../services/attention.js"; +import { assertBoard, assertCompanyAccess } from "./authz.js"; + +export function attentionRoutes(db: Db) { + const router = Router(); + const svc = attentionService(db); + + router.get("/companies/:companyId/attention", async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + assertBoard(req); + if (!req.actor.userId) { + res.status(403).json({ error: "Board user context required" }); + return; + } + + const includeDismissed = req.query.includeDismissed === "true"; + const feed = await svc.list(companyId, { + userId: req.actor.userId, + includeDismissed, + }); + res.json(feed); + }); + + return router; +} diff --git a/server/src/routes/inbox-dismissals.ts b/server/src/routes/inbox-dismissals.ts index 1b51633e56f..1557a2b3cdf 100644 --- a/server/src/routes/inbox-dismissals.ts +++ b/server/src/routes/inbox-dismissals.ts @@ -6,7 +6,7 @@ import { assertCompanyAccess, getActorInfo } from "./authz.js"; import { inboxDismissalService, logActivity } from "../services/index.js"; const inboxDismissalSchema = z.object({ - itemKey: z.string().trim().min(1).regex(/^(approval|join|run):.+$/, "Unsupported inbox item key"), + itemKey: z.string().trim().min(1).regex(/^(approval|join|run|attention):.+$/, "Unsupported inbox item key"), }); export function inboxDismissalRoutes(db: Db) { diff --git a/server/src/routes/index.ts b/server/src/routes/index.ts index c637677c9f1..aef7f589013 100644 --- a/server/src/routes/index.ts +++ b/server/src/routes/index.ts @@ -15,6 +15,7 @@ export { secretRoutes } from "./secrets.js"; export { costRoutes } from "./costs.js"; export { activityRoutes } from "./activity.js"; export { dashboardRoutes } from "./dashboard.js"; +export { attentionRoutes } from "./attention.js"; export { sidebarBadgeRoutes } from "./sidebar-badges.js"; export { sidebarPreferenceRoutes } from "./sidebar-preferences.js"; export { resourceMembershipRoutes } from "./resource-memberships.js"; diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 5f897a14f89..5f3a105d475 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -2813,6 +2813,15 @@ registry.registerPath({ responses: { 200: r.ok(), 401: r.unauthorized }, }); +registry.registerPath({ + method: "get", + path: "/api/companies/{companyId}/attention", + tags: ["inbox"], + summary: "List decision-only attention feed items", + request: { params: z.object({ companyId: z.string() }) }, + responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, +}); + registry.registerPath({ method: "get", path: "/api/sidebar-preferences/me", @@ -2870,7 +2879,7 @@ registry.registerPath({ request: { params: z.object({ companyId: z.string() }), body: jsonBody(z.object({ - itemKey: z.string().trim().min(1).regex(/^(approval|join|run):.+$/, "Unsupported inbox item key"), + itemKey: z.string().trim().min(1).regex(/^(approval|join|run|attention):.+$/, "Unsupported inbox item key"), })), }, responses: { 200: r.ok(), 401: r.unauthorized }, diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts new file mode 100644 index 00000000000..ef240dedb9b --- /dev/null +++ b/server/src/services/attention.ts @@ -0,0 +1,843 @@ +import { and, desc, eq, gt, inArray, isNotNull, isNull, notInArray, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + agents, + approvals, + companies, + heartbeatRunEvents, + heartbeatRuns, + inboxDismissals, + invites, + issueApprovals, + issueRecoveryActions, + issueThreadInteractions, + issues, + joinRequests, +} from "@paperclipai/db"; +import type { + AttentionDecisionVerb, + AttentionFeed, + AttentionItem, + AttentionSeverity, + AttentionSourceKind, + AttentionSubject, +} from "@paperclipai/shared"; +import { PRODUCTIVITY_REVIEW_ORIGIN_KIND } from "./productivity-review.js"; +import { budgetService } from "./budgets.js"; +import { issueService } from "./issues.js"; +import { parseIssueExecutionState } from "./issue-execution-policy.js"; + +const ATTENTION_SOURCE_KINDS: AttentionSourceKind[] = [ + "approval", + "issue_thread_interaction", + "join_request", + "recovery_action", + "productivity_review", + "blocker_attention", + "review", + "failed_run", + "budget_alert", + "agent_error_alert", +]; + +const SEVERITY_RANK: Record = { + critical: 0, + high: 1, + medium: 2, + low: 3, +}; + +const SOURCE_RANK: Record = { + failed_run: 0, + recovery_action: 1, + blocker_attention: 2, + budget_alert: 3, + agent_error_alert: 4, + approval: 5, + issue_thread_interaction: 6, + review: 7, + productivity_review: 8, + join_request: 9, +}; + +const PENDING_INTERACTION_STATUSES = ["pending"] as const; +const OPEN_RECOVERY_STATUSES = ["active", "escalated"] as const; +const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const; +const PRODUCTIVITY_REVIEW_TERMINAL_STATUSES = ["done", "cancelled"] as const; +const FAILED_RUN_STATUSES = ["failed", "timed_out"] as const; + +type IssueSummaryRow = { + id: string; + companyId: string; + identifier: string | null; + title: string; + status: string; + priority: string; + assigneeAgentId: string | null; + assigneeUserId: string | null; + createdAt: Date; + updatedAt: Date; +}; + +type AttentionListOptions = { + userId?: string | null; + includeDismissed?: boolean; +}; + +function emptyCounts(): Record { + return Object.fromEntries(ATTENTION_SOURCE_KINDS.map((kind) => [kind, 0])) as Record; +} + +function toIso(value: Date | string | null | undefined): string { + if (!value) return new Date(0).toISOString(); + return value instanceof Date ? value.toISOString() : new Date(value).toISOString(); +} + +function timestamp(value: Date | string | null | undefined): number { + if (!value) return 0; + const time = new Date(value).getTime(); + return Number.isFinite(time) ? time : 0; +} + +function isDismissed( + dismissedAtByKey: ReadonlyMap, + dismissalKey: string, + activityAt: string, +) { + const dismissedAt = dismissedAtByKey.get(dismissalKey); + return dismissedAt != null && dismissedAt >= timestamp(activityAt); +} + +function issueHref(prefix: string, issue: Pick) { + return `/${prefix}/issues/${issue.identifier ?? issue.id}`; +} + +function issueSubject(prefix: string, issue: IssueSummaryRow): AttentionSubject { + return { + kind: "issue", + id: issue.id, + companyId: issue.companyId, + title: issue.title, + identifier: issue.identifier, + status: issue.status, + href: issueHref(prefix, issue), + metadata: { + priority: issue.priority, + assigneeAgentId: issue.assigneeAgentId, + assigneeUserId: issue.assigneeUserId, + }, + }; +} + +function itemId(sourceKind: AttentionSourceKind, dedupKey: string) { + return `${sourceKind}:${dedupKey}`; +} + +function decisionVerbs(...verbs: AttentionDecisionVerb[]): AttentionDecisionVerb[] { + return verbs; +} + +function createItem(input: Omit): AttentionItem { + return { + ...input, + id: itemId(input.sourceKind, input.dedupKey), + dismissalKey: `attention:${input.dedupKey}`, + rank: 0, + }; +} + +function compareAttentionItems(left: AttentionItem, right: AttentionItem) { + const severityDiff = SEVERITY_RANK[left.severity] - SEVERITY_RANK[right.severity]; + if (severityDiff !== 0) return severityDiff; + const sourceDiff = SOURCE_RANK[left.sourceKind] - SOURCE_RANK[right.sourceKind]; + if (sourceDiff !== 0) return sourceDiff; + const timeDiff = timestamp(right.activityAt) - timestamp(left.activityAt); + if (timeDiff !== 0) return timeDiff; + return left.dedupKey.localeCompare(right.dedupKey); +} + +function betterDuplicate(left: AttentionItem, right: AttentionItem) { + return compareAttentionItems(left, right) <= 0 ? left : right; +} + +function approvalTitle(type: string, payload: Record) { + const title = typeof payload.title === "string" ? payload.title.trim() : ""; + if (title) return title; + const summary = typeof payload.summary === "string" ? payload.summary.trim() : ""; + if (summary) return summary; + return type.replaceAll("_", " "); +} + +function interactionLabel(kind: string) { + switch (kind) { + case "request_confirmation": + return "Confirmation requested"; + case "request_checkbox_confirmation": + return "Selection confirmation requested"; + case "ask_user_questions": + return "Questions need answers"; + case "suggest_tasks": + return "Suggested tasks need a decision"; + default: + return "Interaction needs a decision"; + } +} + +function interactionVerbs(kind: string) { + if (kind === "ask_user_questions") { + return decisionVerbs({ + id: "respond", + label: "Respond", + description: "Submit answers to the pending questions.", + }); + } + return decisionVerbs( + { + id: "accept", + label: "Accept", + description: "Accept the pending interaction.", + }, + { + id: "reject", + label: "Reject", + description: "Reject the pending interaction and provide a reason when required.", + }, + ); +} + +function budgetObservedPercent(amountObserved: number, amountLimit: number) { + return amountLimit > 0 ? Math.round((amountObserved / amountLimit) * 10_000) / 100 : 0; +} + +async function companyPrefix(db: Db, companyId: string) { + const row = await db + .select({ issuePrefix: companies.issuePrefix }) + .from(companies) + .where(eq(companies.id, companyId)) + .then((rows) => rows[0] ?? null); + return row?.issuePrefix ?? "PAP"; +} + +async function dismissedAtByKey(db: Db, companyId: string, userId: string | null | undefined) { + if (!userId) return new Map(); + const rows = await db + .select({ itemKey: inboxDismissals.itemKey, dismissedAt: inboxDismissals.dismissedAt }) + .from(inboxDismissals) + .where(and(eq(inboxDismissals.companyId, companyId), eq(inboxDismissals.userId, userId))); + return new Map(rows.map((row) => [row.itemKey, timestamp(row.dismissedAt)])); +} + +async function issueSummaryMap(db: Db, companyId: string, issueIds: Array) { + const ids = [...new Set(issueIds.filter((value): value is string => Boolean(value)))]; + if (ids.length === 0) return new Map(); + const rows = await db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + createdAt: issues.createdAt, + updatedAt: issues.updatedAt, + }) + .from(issues) + .where(and(eq(issues.companyId, companyId), inArray(issues.id, ids), isNull(issues.hiddenAt))); + return new Map(rows.map((row) => [row.id, row])); +} + +function readRunIssueId(contextSnapshot: Record | null) { + const issueId = contextSnapshot?.issueId ?? contextSnapshot?.taskId; + return typeof issueId === "string" && issueId.length > 0 ? issueId : null; +} + +export function attentionService(db: Db) { + return { + list: async (companyId: string, options: AttentionListOptions = {}): Promise => { + const prefix = await companyPrefix(db, companyId); + const dismissed = await dismissedAtByKey(db, companyId, options.userId); + const includeDismissed = options.includeDismissed === true; + const collected: AttentionItem[] = []; + + const add = (item: AttentionItem) => { + if (!includeDismissed && isDismissed(dismissed, item.dismissalKey, item.activityAt)) return; + collected.push(item); + }; + + const pendingApprovals = await db + .select({ + id: approvals.id, + type: approvals.type, + status: approvals.status, + requestedByAgentId: approvals.requestedByAgentId, + requestedByUserId: approvals.requestedByUserId, + payload: approvals.payload, + createdAt: approvals.createdAt, + updatedAt: approvals.updatedAt, + }) + .from(approvals) + .where(and(eq(approvals.companyId, companyId), eq(approvals.status, "pending"))) + .orderBy(desc(approvals.updatedAt), desc(approvals.id)); + + for (const approval of pendingApprovals) { + const dedupKey = `approval:${approval.id}`; + const title = approvalTitle(approval.type, approval.payload); + add(createItem({ + companyId, + sourceKind: "approval", + subject: { + kind: "approval", + id: approval.id, + companyId, + title, + identifier: null, + status: approval.status, + href: `/${prefix}/approvals/${approval.id}`, + metadata: { + type: approval.type, + requestedByAgentId: approval.requestedByAgentId, + requestedByUserId: approval.requestedByUserId, + }, + }, + whyNow: "Approval is pending a board decision.", + decisionVerbs: decisionVerbs( + { id: "approve", label: "Approve", description: "Approve the request." }, + { id: "reject", label: "Reject", description: "Reject the request." }, + { id: "request_revision", label: "Request revision", description: "Send the request back for changes." }, + ), + inlineResolvable: approval.type !== "request_board_approval", + entryRule: "approvals.status = 'pending'", + exitRule: "Approval leaves pending status.", + dedupKey, + severity: "medium", + activityAt: toIso(approval.updatedAt), + createdAt: toIso(approval.createdAt), + updatedAt: toIso(approval.updatedAt), + relatedIssue: null, + })); + } + + const interactionRows = await db + .select({ + id: issueThreadInteractions.id, + issueId: issueThreadInteractions.issueId, + kind: issueThreadInteractions.kind, + status: issueThreadInteractions.status, + title: issueThreadInteractions.title, + summary: issueThreadInteractions.summary, + createdAt: issueThreadInteractions.createdAt, + updatedAt: issueThreadInteractions.updatedAt, + }) + .from(issueThreadInteractions) + .where(and( + eq(issueThreadInteractions.companyId, companyId), + inArray(issueThreadInteractions.status, [...PENDING_INTERACTION_STATUSES]), + )) + .orderBy(desc(issueThreadInteractions.updatedAt), desc(issueThreadInteractions.id)); + const interactionIssueMap = await issueSummaryMap(db, companyId, interactionRows.map((row) => row.issueId)); + + for (const interaction of interactionRows) { + const issue = interactionIssueMap.get(interaction.issueId) ?? null; + const dedupKey = `interaction:${interaction.id}`; + add(createItem({ + companyId, + sourceKind: "issue_thread_interaction", + subject: { + kind: "interaction", + id: interaction.id, + companyId, + title: interaction.title ?? interaction.summary ?? interactionLabel(interaction.kind), + identifier: null, + status: interaction.status, + href: issue ? `${issueHref(prefix, issue)}#interaction-${interaction.id}` : null, + metadata: { kind: interaction.kind, issueId: interaction.issueId }, + }, + whyNow: `${interactionLabel(interaction.kind)} on an issue thread.`, + decisionVerbs: interactionVerbs(interaction.kind), + inlineResolvable: true, + entryRule: "issue_thread_interactions.status = 'pending'", + exitRule: "Interaction resolves, expires, fails, or is cancelled.", + dedupKey, + severity: "medium", + activityAt: toIso(interaction.updatedAt), + createdAt: toIso(interaction.createdAt), + updatedAt: toIso(interaction.updatedAt), + relatedIssue: issue ? issueSubject(prefix, issue) : null, + })); + } + + const pendingJoins = await db + .select({ + id: joinRequests.id, + requestType: joinRequests.requestType, + status: joinRequests.status, + requestingUserId: joinRequests.requestingUserId, + requestEmailSnapshot: joinRequests.requestEmailSnapshot, + agentName: joinRequests.agentName, + adapterType: joinRequests.adapterType, + createdAt: joinRequests.createdAt, + updatedAt: joinRequests.updatedAt, + }) + .from(joinRequests) + .innerJoin(invites, eq(joinRequests.inviteId, invites.id)) + .where(and( + eq(joinRequests.companyId, companyId), + eq(invites.companyId, companyId), + eq(joinRequests.status, "pending_approval"), + )) + .orderBy(desc(joinRequests.updatedAt), desc(joinRequests.id)); + + for (const join of pendingJoins) { + const label = join.requestType === "agent" + ? join.agentName ?? "Agent join request" + : join.requestEmailSnapshot ?? join.requestingUserId ?? "Human join request"; + const dedupKey = `join:${join.id}`; + add(createItem({ + companyId, + sourceKind: "join_request", + subject: { + kind: "join_request", + id: join.id, + companyId, + title: label, + identifier: null, + status: join.status, + href: `/${prefix}/settings/access`, + metadata: { + requestType: join.requestType, + requestingUserId: join.requestingUserId, + adapterType: join.adapterType, + }, + }, + whyNow: "Join request is pending approval.", + decisionVerbs: decisionVerbs( + { id: "approve", label: "Approve", description: "Approve this join request." }, + { id: "reject", label: "Reject", description: "Reject this join request." }, + ), + inlineResolvable: true, + entryRule: "join_requests.status = 'pending_approval'", + exitRule: "Join request is approved or rejected.", + dedupKey, + severity: "medium", + activityAt: toIso(join.updatedAt), + createdAt: toIso(join.createdAt), + updatedAt: toIso(join.updatedAt), + relatedIssue: null, + })); + } + + const recoveryRows = await db + .select() + .from(issueRecoveryActions) + .where(and( + eq(issueRecoveryActions.companyId, companyId), + inArray(issueRecoveryActions.status, [...OPEN_RECOVERY_STATUSES]), + inArray(issueRecoveryActions.ownerType, [...HUMAN_RECOVERY_OWNER_TYPES]), + )) + .orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id)); + const recoveryIssueMap = await issueSummaryMap( + db, + companyId, + recoveryRows.flatMap((row) => [row.sourceIssueId, row.recoveryIssueId]), + ); + + for (const recovery of recoveryRows) { + const sourceIssue = recoveryIssueMap.get(recovery.sourceIssueId) ?? null; + const recoveryIssue = recovery.recoveryIssueId ? recoveryIssueMap.get(recovery.recoveryIssueId) ?? null : null; + const dedupKey = `recovery:${recovery.kind}:${recovery.sourceIssueId}:${recovery.cause}:${recovery.fingerprint}`; + add(createItem({ + companyId, + sourceKind: "recovery_action", + subject: { + kind: "recovery_action", + id: recovery.id, + companyId, + title: recovery.nextAction, + identifier: null, + status: recovery.status, + href: recoveryIssue ? issueHref(prefix, recoveryIssue) : sourceIssue ? issueHref(prefix, sourceIssue) : null, + metadata: { + kind: recovery.kind, + cause: recovery.cause, + ownerType: recovery.ownerType, + ownerUserId: recovery.ownerUserId, + sourceIssueId: recovery.sourceIssueId, + recoveryIssueId: recovery.recoveryIssueId, + }, + }, + whyNow: recovery.status === "escalated" + ? "Recovery action escalated to a human owner." + : "Recovery action is assigned to a human owner.", + decisionVerbs: decisionVerbs( + { id: "resolve", label: "Resolve", description: "Record the recovery outcome." }, + { id: "reassign", label: "Reassign", description: "Move the recovery to another owner." }, + { id: "cancel", label: "Cancel", description: "Cancel the recovery action." }, + ), + inlineResolvable: false, + entryRule: "issue_recovery_actions.status in ('active','escalated') and owner_type in ('user','board')", + exitRule: "Recovery action resolves, is cancelled, or moves back to an agent/system owner.", + dedupKey, + severity: recovery.status === "escalated" ? "high" : "medium", + activityAt: toIso(recovery.updatedAt), + createdAt: toIso(recovery.createdAt), + updatedAt: toIso(recovery.updatedAt), + relatedIssue: sourceIssue ? issueSubject(prefix, sourceIssue) : null, + })); + } + + const productivityRows = await db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + priority: issues.priority, + originId: issues.originId, + originFingerprint: issues.originFingerprint, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + createdAt: issues.createdAt, + updatedAt: issues.updatedAt, + }) + .from(issues) + .where(and( + eq(issues.companyId, companyId), + eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND), + isNull(issues.hiddenAt), + isNotNull(issues.assigneeUserId), + notInArray(issues.status, [...PRODUCTIVITY_REVIEW_TERMINAL_STATUSES]), + )) + .orderBy(desc(issues.updatedAt), desc(issues.id)); + const productivitySourceMap = await issueSummaryMap(db, companyId, productivityRows.map((row) => row.originId)); + + for (const review of productivityRows) { + const reviewIssue: IssueSummaryRow = review; + const sourceIssue = review.originId ? productivitySourceMap.get(review.originId) ?? null : null; + const dedupKey = `productivity_review:${review.originFingerprint ?? review.originId ?? review.id}`; + add(createItem({ + companyId, + sourceKind: "productivity_review", + subject: issueSubject(prefix, reviewIssue), + whyNow: "Productivity review is awaiting a human decision.", + decisionVerbs: decisionVerbs( + { id: "resolve", label: "Resolve", description: "Record a productivity review outcome." }, + { id: "dismiss", label: "Dismiss", description: "Dismiss this review for now." }, + { id: "reassign", label: "Reassign", description: "Move the review to another owner." }, + ), + inlineResolvable: false, + entryRule: "Open issue_productivity_review issue assigned to a user.", + exitRule: "Review issue is done/cancelled or no longer assigned to a user.", + dedupKey, + severity: review.priority === "critical" ? "critical" : review.priority === "high" ? "high" : "medium", + activityAt: toIso(review.updatedAt), + createdAt: toIso(review.createdAt), + updatedAt: toIso(review.updatedAt), + relatedIssue: sourceIssue ? issueSubject(prefix, sourceIssue) : null, + })); + } + + const blockedIssues = await issueService(db).list(companyId, { status: "blocked", includeBlockedBy: true }); + for (const issue of blockedIssues as Array) { + const blockerAttention = issue.blockerAttention; + if (blockerAttention?.state !== "stalled") continue; + const sample = blockerAttention.sampleStalledBlockerIdentifier ?? blockerAttention.sampleBlockerIdentifier ?? issue.identifier ?? issue.id; + const dedupKey = `blocker:${issue.id}:${sample}`; + add(createItem({ + companyId, + sourceKind: "blocker_attention", + subject: issueSubject(prefix, issue), + whyNow: "Blocked dependency chain is stalled and needs a human to choose the next owner or action.", + decisionVerbs: decisionVerbs( + { id: "unblock", label: "Unblock", description: "Repair or replace the stalled blocker path." }, + { id: "reassign", label: "Reassign", description: "Assign the stalled blocker to a live owner." }, + { id: "nudge", label: "Nudge", description: "Wake or prompt the current owner." }, + ), + inlineResolvable: false, + entryRule: "blocked issue has blockerAttention.state = 'stalled'", + exitRule: "Blocker chain is no longer stalled or the issue leaves blocked status.", + dedupKey, + severity: "high", + activityAt: toIso(issue.updatedAt), + createdAt: toIso(issue.createdAt), + updatedAt: toIso(issue.updatedAt), + relatedIssue: null, + })); + } + + const reviewRows = await db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + executionState: issues.executionState, + createdAt: issues.createdAt, + updatedAt: issues.updatedAt, + }) + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.status, "in_review"), isNull(issues.hiddenAt))) + .orderBy(desc(issues.updatedAt), desc(issues.id)); + const reviewIssueIds = reviewRows.map((row) => row.id); + const pendingReviewApprovalRows = reviewIssueIds.length === 0 + ? [] + : await db + .select({ issueId: issueApprovals.issueId, approvalId: approvals.id }) + .from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)) + .where(and( + eq(issueApprovals.companyId, companyId), + eq(approvals.companyId, companyId), + inArray(issueApprovals.issueId, reviewIssueIds), + eq(approvals.status, "pending"), + )); + const pendingApprovalByIssueId = new Map(pendingReviewApprovalRows.map((row) => [row.issueId, row.approvalId])); + + for (const review of reviewRows) { + const state = parseIssueExecutionState(review.executionState); + const currentParticipant = state?.status === "pending" ? state.currentParticipant : null; + const hasHumanParticipant = currentParticipant?.type === "user"; + const pendingApprovalId = pendingApprovalByIssueId.get(review.id) ?? null; + if (!hasHumanParticipant && !review.assigneeUserId && !pendingApprovalId) continue; + const issue: IssueSummaryRow = review; + const dedupKey = `review:${review.id}`; + add(createItem({ + companyId, + sourceKind: "review", + subject: issueSubject(prefix, issue), + whyNow: pendingApprovalId + ? "Issue is in review with a linked pending approval." + : hasHumanParticipant + ? "Issue is in review and the current execution participant is a user." + : "Issue is in review and assigned to a user.", + decisionVerbs: decisionVerbs( + { id: "approve", label: "Approve", description: "Approve the review and advance the issue." }, + { id: "request_changes", label: "Request changes", description: "Return the issue to the assignee with changes requested." }, + ), + inlineResolvable: false, + entryRule: "issues.status = 'in_review' and human reviewer, user assignee, or linked pending approval exists.", + exitRule: "Issue leaves in_review or the human review path resolves.", + dedupKey, + severity: "medium", + activityAt: toIso(review.updatedAt), + createdAt: toIso(review.createdAt), + updatedAt: toIso(review.updatedAt), + relatedIssue: null, + })); + } + + const exhaustedRunRows = await db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + agentName: agents.name, + status: heartbeatRuns.status, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + contextSnapshot: heartbeatRuns.contextSnapshot, + createdAt: heartbeatRuns.createdAt, + updatedAt: heartbeatRuns.updatedAt, + finishedAt: heartbeatRuns.finishedAt, + exhaustionMessage: heartbeatRunEvents.message, + }) + .from(heartbeatRuns) + .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) + .innerJoin(heartbeatRunEvents, eq(heartbeatRunEvents.runId, heartbeatRuns.id)) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(agents.companyId, companyId), + notInArray(agents.status, ["terminated"]), + inArray(heartbeatRuns.status, [...FAILED_RUN_STATUSES]), + eq(heartbeatRunEvents.companyId, companyId), + eq(heartbeatRunEvents.eventType, "lifecycle"), + sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, + )) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRunEvents.id)); + + const latestExhaustedByRunId = new Map(); + for (const row of exhaustedRunRows) { + if (!latestExhaustedByRunId.has(row.id)) latestExhaustedByRunId.set(row.id, row); + } + const failedRows = [...latestExhaustedByRunId.values()]; + const failedIssueMap = await issueSummaryMap( + db, + companyId, + failedRows.map((row) => readRunIssueId(row.contextSnapshot)), + ); + for (const run of failedRows) { + const issueId = readRunIssueId(run.contextSnapshot); + const newer = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, run.agentId), + gt(heartbeatRuns.createdAt, run.createdAt), + sql`coalesce(${heartbeatRuns.contextSnapshot} ->> 'issueId', ${heartbeatRuns.contextSnapshot} ->> 'taskId', '') = ${issueId ?? ""}`, + )) + .limit(1); + if (newer.length > 0) continue; + + const issue = issueId ? failedIssueMap.get(issueId) ?? null : null; + const dedupKey = `run:${run.id}`; + add(createItem({ + companyId, + sourceKind: "failed_run", + subject: { + kind: "run", + id: run.id, + companyId, + title: `${run.agentName} run ${run.status}`, + identifier: null, + status: run.status, + href: `/${prefix}/agents/${run.agentId}/runs/${run.id}`, + metadata: { + agentId: run.agentId, + agentName: run.agentName, + issueId, + errorCode: run.errorCode, + error: run.error, + retryExhaustedReason: run.exhaustionMessage, + }, + }, + whyNow: "Run failed after automatic retries were exhausted.", + decisionVerbs: decisionVerbs( + { id: "retry", label: "Retry", description: "Retry the failed run or issue." }, + { id: "reassign", label: "Reassign", description: "Move the work to another owner." }, + { id: "dismiss", label: "Dismiss", description: "Dismiss this failed-run attention row." }, + ), + inlineResolvable: true, + entryRule: "latest failed/timed_out run has a Bounded retry exhausted lifecycle event.", + exitRule: "A newer run exists for the same issue/agent pair or the row is dismissed.", + dedupKey, + severity: "high", + activityAt: toIso(run.finishedAt ?? run.updatedAt ?? run.createdAt), + createdAt: toIso(run.createdAt), + updatedAt: toIso(run.updatedAt), + relatedIssue: issue ? issueSubject(prefix, issue) : null, + })); + } + + const budgetOverview = await budgetService(db).overview(companyId); + for (const incident of budgetOverview.activeIncidents) { + const observedPercent = budgetObservedPercent(incident.amountObserved, incident.amountLimit); + if (incident.thresholdType !== "hard" && observedPercent < 85) continue; + const dedupKey = `budget:${incident.policyId}:${toIso(incident.windowStart)}:${incident.thresholdType}`; + add(createItem({ + companyId, + sourceKind: "budget_alert", + subject: { + kind: "budget_incident", + id: incident.id, + companyId, + title: `${incident.scopeName} budget ${incident.thresholdType === "hard" ? "hard stop" : "warning"}`, + identifier: null, + status: incident.status, + href: `/${prefix}/costs`, + metadata: { + policyId: incident.policyId, + scopeType: incident.scopeType, + scopeId: incident.scopeId, + thresholdType: incident.thresholdType, + amountObserved: incident.amountObserved, + amountLimit: incident.amountLimit, + observedPercent, + approvalId: incident.approvalId, + approvalStatus: incident.approvalStatus, + }, + }, + whyNow: incident.thresholdType === "hard" + ? "Budget hard stop was reached." + : "Budget crossed the 85% warning threshold.", + decisionVerbs: decisionVerbs( + { id: "raise_budget_and_resume", label: "Raise budget", description: "Raise the budget and resume paused work." }, + { id: "keep_paused", label: "Keep paused", description: "Dismiss or keep the budget stop in place." }, + ), + inlineResolvable: true, + entryRule: "open budget incident is hard, or soft with observed spend >= 85% of limit.", + exitRule: "Budget incident is resolved or dismissed.", + dedupKey, + severity: incident.thresholdType === "hard" ? "high" : "medium", + activityAt: toIso(incident.updatedAt), + createdAt: toIso(incident.createdAt), + updatedAt: toIso(incident.updatedAt), + relatedIssue: null, + })); + } + + const erroredAgents = await db + .select({ + id: agents.id, + companyId: agents.companyId, + name: agents.name, + role: agents.role, + status: agents.status, + errorReason: agents.errorReason, + createdAt: agents.createdAt, + updatedAt: agents.updatedAt, + }) + .from(agents) + .where(and(eq(agents.companyId, companyId), eq(agents.status, "error"))) + .orderBy(desc(agents.updatedAt), desc(agents.id)); + + for (const agent of erroredAgents) { + const dedupKey = `agent_error:${agent.id}`; + add(createItem({ + companyId, + sourceKind: "agent_error_alert", + subject: { + kind: "agent", + id: agent.id, + companyId, + title: agent.name, + identifier: null, + status: agent.status, + href: `/${prefix}/agents/${agent.id}`, + metadata: { role: agent.role, errorReason: agent.errorReason }, + }, + whyNow: "Agent is in error status and needs operator action or dismissal.", + decisionVerbs: decisionVerbs( + { id: "inspect", label: "Inspect", description: "Inspect the agent error." }, + { id: "dismiss", label: "Dismiss", description: "Dismiss this alert." }, + ), + inlineResolvable: true, + entryRule: "agents.status = 'error'", + exitRule: "Agent leaves error status or the row is dismissed.", + dedupKey, + severity: "high", + activityAt: toIso(agent.updatedAt), + createdAt: toIso(agent.createdAt), + updatedAt: toIso(agent.updatedAt), + relatedIssue: null, + })); + } + + const deduped = new Map(); + for (const item of collected) { + const current = deduped.get(item.dedupKey); + deduped.set(item.dedupKey, current ? betterDuplicate(current, item) : item); + } + + const items = [...deduped.values()] + .sort(compareAttentionItems) + .map((item, index) => ({ ...item, rank: index + 1 })); + const countsBySourceKind = emptyCounts(); + for (const item of items) countsBySourceKind[item.sourceKind] += 1; + + return { + companyId, + generatedAt: new Date().toISOString(), + totalCount: items.length, + countsBySourceKind, + items, + }; + }, + }; +} diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 43524a5642d..a1e15e48e40 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -58,6 +58,7 @@ export { export { goalService } from "./goals.js"; export { activityService, type ActivityFilters } from "./activity.js"; export { workTimelineService, normalizeTimelineWindow } from "./work-timeline.js"; +export { attentionService } from "./attention.js"; export type { WorkTimelineActor, WorkTimelineEdge,