diff --git a/server/src/__tests__/activity-service.test.ts b/server/src/__tests__/activity-service.test.ts index 6a3a117eced..652ba1c076d 100644 --- a/server/src/__tests__/activity-service.test.ts +++ b/server/src/__tests__/activity-service.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; -import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db"; +import { agents, companies, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -26,6 +26,8 @@ describeEmbeddedPostgres("activity service", () => { }, 20_000); afterEach(async () => { + await db.delete(issueComments); + await db.delete(issues); await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(companies); @@ -131,4 +133,89 @@ describeEmbeddedPostgres("activity service", () => { nextAction: "Review the completed output.", }); }); + + it("backfills missing liveness for completed issue runs before returning the ledger", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + const completedAt = new Date("2026-04-18T20:04:00.000Z"); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Fix run ledger", + description: "Make the run ledger answer whether a run advanced.", + status: "done", + priority: "medium", + assigneeAgentId: agentId, + completedAt, + }); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + status: "succeeded", + startedAt: new Date("2026-04-18T20:00:00.000Z"), + finishedAt: completedAt, + contextSnapshot: { issueId }, + resultJson: { + summary: "Finished the implementation.", + }, + livenessState: null, + livenessReason: null, + lastUsefulActionAt: null, + nextAction: null, + }); + + await db.insert(issueComments).values({ + companyId, + issueId, + authorAgentId: agentId, + createdByRunId: runId, + body: "Done", + createdAt: completedAt, + }); + + const runs = await activityService(db).runsForIssue(companyId, issueId); + + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ + runId, + livenessState: "completed", + livenessReason: "Issue is done", + continuationAttempt: 0, + lastUsefulActionAt: completedAt, + }); + + const [persisted] = await db.select().from(heartbeatRuns); + expect(persisted).toMatchObject({ + id: runId, + livenessState: "completed", + livenessReason: "Issue is done", + continuationAttempt: 0, + lastUsefulActionAt: completedAt, + }); + }); }); diff --git a/server/src/services/activity.ts b/server/src/services/activity.ts index cd9f3c1c08e..08e417b8d04 100644 --- a/server/src/services/activity.ts +++ b/server/src/services/activity.ts @@ -1,6 +1,17 @@ import { and, desc, eq, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { activityLog, agents, heartbeatRuns, issues } from "@paperclipai/db"; +import { + activityLog, + agents, + heartbeatRunEvents, + heartbeatRuns, + issueComments, + issueDocuments, + issues, + issueWorkProducts, + workspaceOperations, +} from "@paperclipai/db"; +import { classifyRunLiveness } from "./run-liveness.js"; export interface ActivityFilters { companyId: string; @@ -85,6 +96,197 @@ export function activityService(db: Db) { end `.as("resultJson"); + function countValue(value: unknown) { + const parsed = Number(value ?? 0); + return Number.isFinite(parsed) ? Math.max(0, Math.floor(parsed)) : 0; + } + + function dateValue(value: unknown) { + if (value instanceof Date) return Number.isNaN(value.getTime()) ? null : value; + if (typeof value === "string" || typeof value === "number") { + const parsed = new Date(value); + return Number.isNaN(parsed.getTime()) ? null : parsed; + } + return null; + } + + function latestDate(...values: unknown[]) { + let latest: Date | null = null; + for (const value of values) { + const parsed = dateValue(value); + if (!parsed) continue; + if (!latest || parsed.getTime() > latest.getTime()) latest = parsed; + } + return latest; + } + + function asRecord(value: unknown): Record | null { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + return value as Record; + } + + function readNumber(value: unknown) { + return typeof value === "number" && Number.isFinite(value) ? value : null; + } + + async function backfillMissingRunLivenessForIssue(companyId: string, issueId: string) { + const runs = await db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + status: heartbeatRuns.status, + contextSnapshot: heartbeatRuns.contextSnapshot, + resultJson: heartbeatRuns.resultJson, + stdoutExcerpt: heartbeatRuns.stdoutExcerpt, + stderrExcerpt: heartbeatRuns.stderrExcerpt, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + continuationAttempt: heartbeatRuns.continuationAttempt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + isNull(heartbeatRuns.livenessState), + sql`${heartbeatRuns.status} not in ('queued', 'running')`, + or( + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + sql`exists ( + select 1 + from ${activityLog} + where ${activityLog.companyId} = ${companyId} + and ${activityLog.entityType} = 'issue' + and ${activityLog.entityId} = ${issueId} + and ${activityLog.runId} = ${heartbeatRuns.id} + )`, + ), + ), + ) + .limit(20); + + if (runs.length === 0) return; + + const issue = await db + .select({ + status: issues.status, + title: issues.title, + description: issues.description, + }) + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))) + .then((rows) => rows[0] ?? null); + + for (const run of runs) { + const context = asRecord(run.contextSnapshot); + const continuationAttempt = + readNumber(context?.continuationAttempt) ?? + readNumber(context?.livenessContinuationAttempt) ?? + run.continuationAttempt ?? + 0; + + const [commentStats] = await db + .select({ + count: sql`count(*)::int`, + latestAt: sql`max(${issueComments.createdAt})`, + }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, companyId), + eq(issueComments.issueId, issueId), + eq(issueComments.createdByRunId, run.id), + ), + ); + + const [documentStats] = await db + .select({ + count: sql`count(*)::int`, + planCount: sql`count(*) filter (where ${issueDocuments.key} = 'plan')::int`, + latestAt: sql`max(${issueDocuments.updatedAt})`, + }) + .from(issueDocuments) + .where(and(eq(issueDocuments.companyId, companyId), eq(issueDocuments.issueId, issueId))); + + const [workProductStats] = await db + .select({ + count: sql`count(*)::int`, + latestAt: sql`max(${issueWorkProducts.createdAt})`, + }) + .from(issueWorkProducts) + .where( + and( + eq(issueWorkProducts.companyId, companyId), + eq(issueWorkProducts.issueId, issueId), + eq(issueWorkProducts.createdByRunId, run.id), + ), + ); + + const [workspaceOperationStats] = await db + .select({ + count: sql`count(*)::int`, + latestAt: sql`max(${workspaceOperations.startedAt})`, + }) + .from(workspaceOperations) + .where(and(eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.heartbeatRunId, run.id))); + + const [activityStats] = await db + .select({ + count: sql`count(*)::int`, + latestAt: sql`max(${activityLog.createdAt})`, + }) + .from(activityLog) + .where(and(eq(activityLog.companyId, companyId), eq(activityLog.runId, run.id))); + + const [eventStats] = await db + .select({ + count: sql`count(*) filter (where ${heartbeatRunEvents.eventType} not in ('lifecycle', 'adapter.invoke', 'error'))::int`, + latestAt: sql`max(${heartbeatRunEvents.createdAt}) filter (where ${heartbeatRunEvents.eventType} not in ('lifecycle', 'adapter.invoke', 'error'))`, + }) + .from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, companyId), eq(heartbeatRunEvents.runId, run.id))); + + const classification = classifyRunLiveness({ + runStatus: run.status, + issue, + resultJson: asRecord(run.resultJson), + stdoutExcerpt: run.stdoutExcerpt, + stderrExcerpt: run.stderrExcerpt, + error: run.error, + errorCode: run.errorCode, + continuationAttempt, + evidence: { + issueCommentsCreated: countValue(commentStats?.count), + documentRevisionsCreated: countValue(documentStats?.count), + planDocumentRevisionsCreated: countValue(documentStats?.planCount), + workProductsCreated: countValue(workProductStats?.count), + workspaceOperationsCreated: countValue(workspaceOperationStats?.count), + activityEventsCreated: countValue(activityStats?.count), + toolOrActionEventsCreated: countValue(eventStats?.count), + latestEvidenceAt: latestDate( + commentStats?.latestAt, + documentStats?.latestAt, + workProductStats?.latestAt, + workspaceOperationStats?.latestAt, + activityStats?.latestAt, + eventStats?.latestAt, + ), + }, + }); + + await db + .update(heartbeatRuns) + .set({ + livenessState: classification.livenessState, + livenessReason: classification.livenessReason, + continuationAttempt: classification.continuationAttempt, + lastUsefulActionAt: classification.lastUsefulActionAt, + nextAction: classification.nextAction, + updatedAt: new Date(), + }) + .where(and(eq(heartbeatRuns.id, run.id), isNull(heartbeatRuns.livenessState))); + } + } + return { list: (filters: ActivityFilters) => { const conditions = [eq(activityLog.companyId, filters.companyId)]; @@ -134,8 +336,9 @@ export function activityService(db: Db) { ) .orderBy(desc(activityLog.createdAt)), - runsForIssue: (companyId: string, issueId: string) => - db + runsForIssue: async (companyId: string, issueId: string) => { + await backfillMissingRunLivenessForIssue(companyId, issueId); + return db .select({ runId: heartbeatRuns.id, status: heartbeatRuns.status, @@ -178,7 +381,8 @@ export function activityService(db: Db) { ), ), ) - .orderBy(desc(heartbeatRuns.createdAt)), + .orderBy(desc(heartbeatRuns.createdAt)); + }, issuesForRun: async (runId: string) => { const run = await db