Backfill liveness for issue run ledgers

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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,
+    });
+  });
 });

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<string, unknown> | null {
+    if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+    return value as Record<string, unknown>;
+  }
+
+  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<number>`count(*)::int`,
+          latestAt: sql<Date | null>`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<number>`count(*)::int`,
+          planCount: sql<number>`count(*) filter (where ${issueDocuments.key} = 'plan')::int`,
+          latestAt: sql<Date | null>`max(${issueDocuments.updatedAt})`,
+        })
+        .from(issueDocuments)
+        .where(and(eq(issueDocuments.companyId, companyId), eq(issueDocuments.issueId, issueId)));
+
+      const [workProductStats] = await db
+        .select({
+          count: sql<number>`count(*)::int`,
+          latestAt: sql<Date | null>`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<number>`count(*)::int`,
+          latestAt: sql<Date | null>`max(${workspaceOperations.startedAt})`,
+        })
+        .from(workspaceOperations)
+        .where(and(eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.heartbeatRunId, run.id)));
+
+      const [activityStats] = await db
+        .select({
+          count: sql<number>`count(*)::int`,
+          latestAt: sql<Date | null>`max(${activityLog.createdAt})`,
+        })
+        .from(activityLog)
+        .where(and(eq(activityLog.companyId, companyId), eq(activityLog.runId, run.id)));
+
+      const [eventStats] = await db
+        .select({
+          count: sql<number>`count(*) filter (where ${heartbeatRunEvents.eventType} not in ('lifecycle', 'adapter.invoke', 'error'))::int`,
+          latestAt: sql<Date | null>`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