Implement productivity review reconciler

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

packages/db/src/migrations/0074_striped_genesis.sql

@@ -0,0 +1,4 @@
+CREATE UNIQUE INDEX "issues_active_productivity_review_uq" ON "issues" USING btree ("company_id","origin_kind","origin_id") WHERE "issues"."origin_kind" = 'issue_productivity_review'
+          and "issues"."origin_id" is not null
+          and "issues"."hidden_at" is null
+          and "issues"."status" not in ('done', 'cancelled');

packages/db/src/migrations/meta/_journal.json

@@ -519,6 +519,13 @@
       "when": 1777382021347,
       "tag": "0073_shiny_salo",
       "breakpoints": true
+    },
+    {
+      "idx": 74,
+      "version": "7",
+      "when": 1777384535070,
+      "tag": "0074_striped_genesis",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/issues.ts

@@ -115,6 +115,14 @@ export const issues = pgTable(
           and ${table.hiddenAt} is null
           and ${table.status} not in ('done', 'cancelled')`,
       ),
+    activeProductivityReviewIdx: uniqueIndex("issues_active_productivity_review_uq")
+      .on(table.companyId, table.originKind, table.originId)
+      .where(
+        sql`${table.originKind} = 'issue_productivity_review'
+          and ${table.originId} is not null
+          and ${table.hiddenAt} is null
+          and ${table.status} not in ('done', 'cancelled')`,
+      ),
     activeStrandedIssueRecoveryIdx: uniqueIndex("issues_active_stranded_issue_recovery_uq")
       .on(table.companyId, table.originKind, table.originId)
       .where(

server/src/__tests__/productivity-review-service.test.ts

@@ -0,0 +1,361 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  activityLog,
+  agents,
+  companies,
+  createDb,
+  heartbeatRuns,
+  issueComments,
+  issues,
+} from "@paperclipai/db";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+import {
+  DEFAULT_PRODUCTIVITY_REVIEW_NO_COMMENT_STREAK_RUNS,
+  PRODUCTIVITY_REVIEW_ORIGIN_KIND,
+  productivityReviewService,
+} from "../services/productivity-review.ts";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres productivity review tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
+
+describeEmbeddedPostgres("productivity review service", () => {
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  let db: ReturnType<typeof createDb>;
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-productivity-review-");
+    db = createDb(tempDb.connectionString);
+  }, 30_000);
+
+  afterEach(async () => {
+    await db.execute(sql.raw(`TRUNCATE TABLE "companies" CASCADE`));
+  });
+
+  afterAll(async () => {
+    await tempDb?.cleanup();
+  });
+
+  async function seedAssignedIssue(opts?: {
+    status?: "todo" | "in_progress";
+    startedAt?: Date;
+    parentId?: string | null;
+    originKind?: string;
+  }) {
+    const companyId = randomUUID();
+    const managerId = randomUUID();
+    const coderId = randomUUID();
+    const issueId = randomUUID();
+    const issuePrefix = `PR${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
+    const createdAt = new Date("2026-04-28T10:00:00.000Z");
+
+    await db.insert(companies).values({
+      id: companyId,
+      name: "Productivity Review Co",
+      issuePrefix,
+      requireBoardApprovalForNewAgents: false,
+    });
+    await db.insert(agents).values([
+      {
+        id: managerId,
+        companyId,
+        name: "CTO",
+        role: "cto",
+        status: "idle",
+        adapterType: "codex_local",
+        adapterConfig: {},
+        runtimeConfig: {},
+        permissions: {},
+      },
+      {
+        id: coderId,
+        companyId,
+        name: "Coder",
+        role: "engineer",
+        status: "idle",
+        reportsTo: managerId,
+        adapterType: "codex_local",
+        adapterConfig: {},
+        runtimeConfig: {},
+        permissions: {},
+      },
+    ]);
+    await db.insert(issues).values({
+      id: issueId,
+      companyId,
+      title: "Implement data import",
+      status: opts?.status ?? "in_progress",
+      priority: "medium",
+      assigneeAgentId: coderId,
+      parentId: opts?.parentId ?? null,
+      originKind: opts?.originKind ?? "manual",
+      issueNumber: 1,
+      identifier: `${issuePrefix}-1`,
+      startedAt: opts?.startedAt ?? createdAt,
+      createdAt,
+      updatedAt: createdAt,
+    });
+
+    return { companyId, managerId, coderId, issueId, issuePrefix, createdAt };
+  }
+
+  async function insertRuns(input: {
+    companyId: string;
+    agentId: string;
+    issueId: string;
+    count: number;
+    now: Date;
+    withRunComments?: boolean;
+  }) {
+    const runs: Array<typeof heartbeatRuns.$inferInsert> = [];
+    for (let index = 0; index < input.count; index += 1) {
+      const runId = randomUUID();
+      const createdAt = new Date(input.now.getTime() - index * 60_000);
+      runs.push({
+        id: runId,
+        companyId: input.companyId,
+        agentId: input.agentId,
+        status: "succeeded",
+        invocationSource: "assignment",
+        triggerDetail: "system",
+        startedAt: createdAt,
+        finishedAt: new Date(createdAt.getTime() + 30_000),
+        contextSnapshot: { issueId: input.issueId, taskId: input.issueId },
+        livenessState: "advanced",
+        nextAction: "Continue processing the next batch.",
+        createdAt,
+        updatedAt: createdAt,
+      });
+    }
+    await db.insert(heartbeatRuns).values(runs);
+
+    if (input.withRunComments) {
+      await db.insert(issueComments).values(
+        runs.map((run, index) => ({
+          companyId: input.companyId,
+          issueId: input.issueId,
+          authorAgentId: input.agentId,
+          createdByRunId: run.id,
+          body: `Progress update ${index}`,
+          createdAt: run.createdAt as Date,
+          updatedAt: run.createdAt as Date,
+        })),
+      );
+    }
+
+    return runs;
+  }
+
+  async function listProductivityReviews(companyId: string) {
+    return db
+      .select()
+      .from(issues)
+      .where(and(eq(issues.companyId, companyId), eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND)))
+      .orderBy(issues.createdAt);
+  }
+
+  it("creates exactly one manager-assigned review for a no-comment run streak and refreshes it idempotently", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue();
+    await insertRuns({
+      companyId: seeded.companyId,
+      agentId: seeded.coderId,
+      issueId: seeded.issueId,
+      count: DEFAULT_PRODUCTIVITY_REVIEW_NO_COMMENT_STREAK_RUNS,
+      now,
+    });
+
+    const service = productivityReviewService(db);
+    const first = await service.reconcileProductivityReviews({ now, companyId: seeded.companyId });
+    const second = await service.reconcileProductivityReviews({ now, companyId: seeded.companyId });
+
+    expect(first.created).toBe(1);
+    expect(second.updated).toBe(1);
+    const reviews = await listProductivityReviews(seeded.companyId);
+    expect(reviews).toHaveLength(1);
+    expect(reviews[0]?.parentId).toBe(seeded.issueId);
+    expect(reviews[0]?.assigneeAgentId).toBe(seeded.managerId);
+    expect(reviews[0]?.originId).toBe(seeded.issueId);
+    expect(reviews[0]?.originFingerprint).toBe(`productivity-review:${seeded.issueId}`);
+    expect(reviews[0]?.description).toContain("Primary trigger: `no_comment_streak`");
+    expect(reviews[0]?.description).toContain("No-comment completed-run streak: 10");
+
+    const comments = await db
+      .select()
+      .from(issueComments)
+      .where(eq(issueComments.issueId, reviews[0]!.id));
+    expect(comments.some((comment) => comment.body.includes("Productivity review evidence refreshed"))).toBe(true);
+  });
+
+  it("creates a long-active review without enabling a continuation hold", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue({
+      status: "in_progress",
+      startedAt: new Date(now.getTime() - 7 * 60 * 60 * 1000),
+    });
+    const service = productivityReviewService(db);
+
+    const result = await service.reconcileProductivityReviews({ now, companyId: seeded.companyId });
+    const hold = await service.isProductivityReviewContinuationHoldActive({
+      companyId: seeded.companyId,
+      issueId: seeded.issueId,
+      agentId: seeded.coderId,
+      now,
+    });
+
+    expect(result.created).toBe(1);
+    const [review] = await listProductivityReviews(seeded.companyId);
+    expect(review?.description).toContain("Primary trigger: `long_active_duration`");
+    expect(review?.priority).toBe("medium");
+    expect(hold.held).toBe(false);
+  });
+
+  it("creates a high-churn review even when every sampled run has a progress comment", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue();
+    await insertRuns({
+      companyId: seeded.companyId,
+      agentId: seeded.coderId,
+      issueId: seeded.issueId,
+      count: 10,
+      now,
+      withRunComments: true,
+    });
+
+    const result = await productivityReviewService(db).reconcileProductivityReviews({
+      now,
+      companyId: seeded.companyId,
+    });
+
+    expect(result.created).toBe(1);
+    const [review] = await listProductivityReviews(seeded.companyId);
+    expect(review?.description).toContain("Primary trigger: `high_churn`");
+    expect(review?.description).toContain("Runs in rolling windows: 10/1h");
+  });
+
+  it("skips productivity-review descendants so reviews cannot recursively spawn reviews", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue();
+    const reviewId = randomUUID();
+    const childId = randomUUID();
+    await db.insert(issues).values({
+      id: reviewId,
+      companyId: seeded.companyId,
+      title: "Existing productivity review",
+      status: "todo",
+      priority: "high",
+      originKind: PRODUCTIVITY_REVIEW_ORIGIN_KIND,
+      originId: seeded.issueId,
+      originFingerprint: `productivity-review:${seeded.issueId}`,
+      parentId: seeded.issueId,
+      issueNumber: 2,
+      identifier: `${seeded.issuePrefix}-2`,
+    });
+    await db.insert(issues).values({
+      id: childId,
+      companyId: seeded.companyId,
+      title: "Review follow-up child",
+      status: "in_progress",
+      priority: "medium",
+      assigneeAgentId: seeded.coderId,
+      parentId: reviewId,
+      issueNumber: 3,
+      identifier: `${seeded.issuePrefix}-3`,
+      startedAt: new Date(now.getTime() - 7 * 60 * 60 * 1000),
+    });
+    await insertRuns({
+      companyId: seeded.companyId,
+      agentId: seeded.coderId,
+      issueId: childId,
+      count: 10,
+      now,
+    });
+
+    const result = await productivityReviewService(db).reconcileProductivityReviews({
+      now,
+      companyId: seeded.companyId,
+    });
+    const reviews = await listProductivityReviews(seeded.companyId);
+
+    expect(result.created).toBe(0);
+    expect(reviews).toHaveLength(1);
+  });
+
+  it("treats a recently completed review as a snooze window", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue();
+    await insertRuns({
+      companyId: seeded.companyId,
+      agentId: seeded.coderId,
+      issueId: seeded.issueId,
+      count: 10,
+      now,
+    });
+    const service = productivityReviewService(db);
+    await service.reconcileProductivityReviews({ now, companyId: seeded.companyId });
+    const [review] = await listProductivityReviews(seeded.companyId);
+    await db
+      .update(issues)
+      .set({ status: "done", updatedAt: now })
+      .where(eq(issues.id, review!.id));
+
+    const result = await service.reconcileProductivityReviews({
+      now: new Date(now.getTime() + 30 * 60 * 1000),
+      companyId: seeded.companyId,
+    });
+    const reviews = await listProductivityReviews(seeded.companyId);
+
+    expect(result.snoozed).toBe(1);
+    expect(reviews).toHaveLength(1);
+  });
+
+  it("reports and logs soft-stop holds for open no-comment reviews", async () => {
+    const now = new Date("2026-04-28T12:00:00.000Z");
+    const seeded = await seedAssignedIssue();
+    const [latestRun] = await insertRuns({
+      companyId: seeded.companyId,
+      agentId: seeded.coderId,
+      issueId: seeded.issueId,
+      count: 10,
+      now,
+    });
+    const service = productivityReviewService(db);
+    await service.reconcileProductivityReviews({ now, companyId: seeded.companyId });
+    const [review] = await listProductivityReviews(seeded.companyId);
+
+    const hold = await service.isProductivityReviewContinuationHoldActive({
+      companyId: seeded.companyId,
+      issueId: seeded.issueId,
+      agentId: seeded.coderId,
+      now,
+    });
+    expect(hold.held).toBe(true);
+    if (!hold.held) return;
+
+    await service.recordContinuationHold({
+      companyId: seeded.companyId,
+      issueId: seeded.issueId,
+      runId: latestRun!.id as string,
+      agentId: seeded.coderId,
+      reviewIssueId: review!.id,
+      trigger: hold.trigger,
+      reason: hold.reason,
+    });
+    const activities = await db
+      .select()
+      .from(activityLog)
+      .where(eq(activityLog.action, "issue.productivity_review_continuation_held"));
+    expect(activities).toHaveLength(1);
+    expect(activities[0]?.entityId).toBe(seeded.issueId);
+  });
+});

server/src/index.ts

@@ -705,6 +705,12 @@ export async function startServer(): Promise<StartedServer> {
           logger.warn({ ...scanned }, "startup active-run output watchdog created review work");
         }
       })
+      .then(async () => {
+        const reviewed = await heartbeat.reconcileProductivityReviews();
+        if (reviewed.created > 0 || reviewed.updated > 0) {
+          logger.warn({ ...reviewed }, "startup productivity reconciliation created or updated review work");
+        }
+      })
       .catch((err) => {
         logger.error({ err }, "startup heartbeat recovery failed");
       });
@@ -764,6 +770,12 @@ export async function startServer(): Promise<StartedServer> {
             logger.warn({ ...scanned }, "periodic active-run output watchdog created review work");
           }
         })
+        .then(async () => {
+          const reviewed = await heartbeat.reconcileProductivityReviews();
+          if (reviewed.created > 0 || reviewed.updated > 0) {
+            logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work");
+          }
+        })
         .catch((err) => {
           logger.error({ err }, "periodic heartbeat recovery failed");
         });

server/src/services/heartbeat.ts

@@ -110,6 +110,7 @@ import {
 } from "./recovery/index.js";
 import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js";
 import { recoveryService } from "./recovery/service.js";
+import { productivityReviewService } from "./productivity-review.js";
 import { withAgentStartLock } from "./agent-start-lock.js";
 import { redactCurrentUserText, redactCurrentUserValue } from "../log-redaction.js";
 import {
@@ -2004,6 +2005,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
   };
   const budgets = budgetService(db, budgetHooks);
   const recovery = recoveryService(db, { enqueueWakeup });
+  const productivityReviews = productivityReviewService(db, { enqueueWakeup });
   let unsafeTextProjectionPromise: Promise<boolean> | null = null;
 
   async function hasUnsafeTextProjectionDatabase() {
@@ -2807,6 +2809,29 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
           projectId: issue.projectId,
         })
         : null;
+    if (issue) {
+      const productivityHold = await productivityReviews.isProductivityReviewContinuationHoldActive({
+        companyId: issue.companyId,
+        issueId: issue.id,
+        agentId: run.agentId,
+      });
+      if (productivityHold.held) {
+        await setRunStatus(run.id, run.status, {
+          livenessReason:
+            `${run.livenessReason ?? "Run ended without concrete progress"}; continuation held by productivity review ${productivityHold.reviewIdentifier ?? productivityHold.reviewIssueId}`,
+        });
+        await productivityReviews.recordContinuationHold({
+          companyId: issue.companyId,
+          issueId: issue.id,
+          runId: run.id,
+          agentId: run.agentId,
+          reviewIssueId: productivityHold.reviewIssueId,
+          trigger: productivityHold.trigger,
+          reason: productivityHold.reason,
+        });
+        return;
+      }
+    }
 
     const nextAttempt = readContinuationAttempt(run.continuationAttempt) + 1;
     const idempotencyKey = issue
@@ -4494,6 +4519,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     return recovery.scanSilentActiveRuns(opts);
   }
 
+  async function reconcileProductivityReviews(opts?: { now?: Date; companyId?: string }) {
+    return productivityReviews.reconcileProductivityReviews(opts);
+  }
+
   async function buildRunOutputSilence(
     run: Pick<
       typeof heartbeatRuns.$inferSelect,
@@ -7494,6 +7523,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
 
     scanSilentActiveRuns,
 
+    reconcileProductivityReviews,
+
     buildRunOutputSilence,
 
     tickTimers: async (now = new Date()) => {

server/src/services/index.ts

@@ -32,6 +32,10 @@ export { routineService } from "./routines.js";
 export { costService } from "./costs.js";
 export { financeService } from "./finance.js";
 export { heartbeatService } from "./heartbeat.js";
+export {
+  productivityReviewService,
+  PRODUCTIVITY_REVIEW_ORIGIN_KIND,
+} from "./productivity-review.js";
 export { classifyIssueGraphLiveness, type IssueLivenessFinding } from "./recovery/index.js";
 export { dashboardService } from "./dashboard.js";
 export { sidebarBadgeService } from "./sidebar-badges.js";

server/src/services/productivity-review.ts

@@ -0,0 +1,758 @@
+import { and, asc, desc, eq, gt, inArray, isNull, notInArray, sql } from "drizzle-orm";
+import type { Db } from "@paperclipai/db";
+import {
+  agents,
+  companies,
+  costEvents,
+  heartbeatRuns,
+  issueComments,
+  issues,
+  projects,
+} from "@paperclipai/db";
+import { logActivity } from "./activity-log.js";
+import { budgetService } from "./budgets.js";
+import { issueService } from "./issues.js";
+import { RECOVERY_ORIGIN_KINDS } from "./recovery/origins.js";
+
+export const PRODUCTIVITY_REVIEW_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.issueProductivityReview;
+export const DEFAULT_PRODUCTIVITY_REVIEW_NO_COMMENT_STREAK_RUNS = 10;
+export const DEFAULT_PRODUCTIVITY_REVIEW_LONG_ACTIVE_HOURS = 6;
+export const DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_HOURLY = 10;
+export const DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_SIX_HOURS = 30;
+export const DEFAULT_PRODUCTIVITY_REVIEW_RESOLVED_SNOOZE_MS = 6 * 60 * 60 * 1000;
+
+const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const;
+const ACTIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
+const MAX_CANDIDATE_ISSUES = 250;
+const MAX_RUNS_FOR_STREAK = 100;
+const MAX_PARENT_WALK_DEPTH = 25;
+
+type IssueRow = typeof issues.$inferSelect;
+type AgentRow = typeof agents.$inferSelect;
+type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
+type ProductivityReviewTrigger = "no_comment_streak" | "long_active_duration" | "high_churn";
+
+type ProductivityReviewThresholds = {
+  noCommentStreakRuns: number;
+  longActiveMs: number;
+  highChurnHourly: number;
+  highChurnSixHours: number;
+  resolvedSnoozeMs: number;
+};
+
+type ProductivityReviewEvidence = {
+  trigger: ProductivityReviewTrigger;
+  triggerReasons: string[];
+  sourceIssue: IssueRow;
+  sourceAgent: AgentRow;
+  noCommentStreak: number;
+  totalRunCount: number;
+  terminalRunCount: number;
+  activeRunCount: number;
+  runCountLastHour: number;
+  runCountLastSixHours: number;
+  commentCount: number;
+  commentCountLastHour: number;
+  commentCountLastSixHours: number;
+  elapsedMs: number | null;
+  latestRuns: HeartbeatRunRow[];
+  latestComments: Array<typeof issueComments.$inferSelect>;
+  costCents: number;
+  usageSamples: Array<{ runId: string; usageJson: Record<string, unknown> | null }>;
+  nextAction: string | null;
+  thresholds: ProductivityReviewThresholds;
+  generatedAt: Date;
+};
+
+type EnqueueWakeup = (
+  agentId: string,
+  opts?: {
+    source?: "timer" | "assignment" | "on_demand" | "automation";
+    triggerDetail?: "manual" | "ping" | "callback" | "system";
+    reason?: string | null;
+    payload?: Record<string, unknown> | null;
+    requestedByActorType?: "user" | "agent" | "system";
+    requestedByActorId?: string | null;
+    contextSnapshot?: Record<string, unknown>;
+  },
+) => Promise<unknown | null>;
+
+function productivityReviewFingerprint(sourceIssueId: string) {
+  return `productivity-review:${sourceIssueId}`;
+}
+
+function issueRunScopeSql(issueId: string) {
+  return sql`(
+    ${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}
+    or ${heartbeatRuns.contextSnapshot}->>'taskId' = ${issueId}
+    or ${heartbeatRuns.contextSnapshot}->>'taskKey' = ${issueId}
+  )`;
+}
+
+function msToHuman(ms: number | null) {
+  if (ms === null) return "unknown";
+  const minutes = Math.floor(ms / 60_000);
+  if (minutes < 60) return `${minutes}m`;
+  const hours = Math.floor(minutes / 60);
+  const days = Math.floor(hours / 24);
+  if (days > 0) return `${days}d ${hours % 24}h`;
+  return `${hours}h ${minutes % 60}m`;
+}
+
+function issueUiLink(issue: { identifier: string | null; id: string }, prefix: string) {
+  const label = issue.identifier ?? issue.id;
+  return `[${label}](/${prefix}/issues/${label})`;
+}
+
+function runUiLink(run: { id: string; agentId: string }, prefix: string) {
+  return `[${run.id}](/${prefix}/agents/${run.agentId}/runs/${run.id})`;
+}
+
+function truncateInline(value: string | null | undefined, max = 260) {
+  if (!value) return "";
+  const compact = value.replace(/\s+/g, " ").trim();
+  return compact.length <= max ? compact : `${compact.slice(0, max - 3)}...`;
+}
+
+function readPositiveInteger(value: number, fallback: number) {
+  return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
+}
+
+function buildThresholds(overrides?: Partial<ProductivityReviewThresholds>): ProductivityReviewThresholds {
+  return {
+    noCommentStreakRuns: readPositiveInteger(
+      overrides?.noCommentStreakRuns ?? DEFAULT_PRODUCTIVITY_REVIEW_NO_COMMENT_STREAK_RUNS,
+      DEFAULT_PRODUCTIVITY_REVIEW_NO_COMMENT_STREAK_RUNS,
+    ),
+    longActiveMs: readPositiveInteger(
+      overrides?.longActiveMs ?? DEFAULT_PRODUCTIVITY_REVIEW_LONG_ACTIVE_HOURS * 60 * 60 * 1000,
+      DEFAULT_PRODUCTIVITY_REVIEW_LONG_ACTIVE_HOURS * 60 * 60 * 1000,
+    ),
+    highChurnHourly: readPositiveInteger(
+      overrides?.highChurnHourly ?? DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_HOURLY,
+      DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_HOURLY,
+    ),
+    highChurnSixHours: readPositiveInteger(
+      overrides?.highChurnSixHours ?? DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_SIX_HOURS,
+      DEFAULT_PRODUCTIVITY_REVIEW_HIGH_CHURN_SIX_HOURS,
+    ),
+    resolvedSnoozeMs: readPositiveInteger(
+      overrides?.resolvedSnoozeMs ?? DEFAULT_PRODUCTIVITY_REVIEW_RESOLVED_SNOOZE_MS,
+      DEFAULT_PRODUCTIVITY_REVIEW_RESOLVED_SNOOZE_MS,
+    ),
+  };
+}
+
+function choosePrimaryTrigger(input: {
+  noComment: boolean;
+  longActive: boolean;
+  highChurn: boolean;
+}): ProductivityReviewTrigger | null {
+  if (input.noComment) return "no_comment_streak";
+  if (input.highChurn) return "high_churn";
+  if (input.longActive) return "long_active_duration";
+  return null;
+}
+
+function isSoftStopTrigger(trigger: ProductivityReviewTrigger) {
+  return trigger === "no_comment_streak" || trigger === "high_churn";
+}
+
+function formatTrigger(trigger: ProductivityReviewTrigger) {
+  if (trigger === "no_comment_streak") return "No-comment streak";
+  if (trigger === "high_churn") return "High churn";
+  return "Long active duration";
+}
+
+export function productivityReviewService(db: Db, deps?: { enqueueWakeup?: EnqueueWakeup }) {
+  const issuesSvc = issueService(db);
+  const budgets = budgetService(db);
+
+  async function getCompanyIssuePrefix(companyId: string) {
+    return db
+      .select({ issuePrefix: companies.issuePrefix })
+      .from(companies)
+      .where(eq(companies.id, companyId))
+      .then((rows) => rows[0]?.issuePrefix ?? "PAP");
+  }
+
+  async function getAgent(agentId: string) {
+    return db
+      .select()
+      .from(agents)
+      .where(eq(agents.id, agentId))
+      .then((rows) => rows[0] ?? null);
+  }
+
+  function isAgentInvokable(agent: AgentRow | null | undefined) {
+    return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status));
+  }
+
+  async function isProductivityReviewDescendant(issue: Pick<IssueRow, "companyId" | "parentId">) {
+    let parentId = issue.parentId;
+    let depth = 0;
+    while (parentId && depth < MAX_PARENT_WALK_DEPTH) {
+      const parent = await db
+        .select({ id: issues.id, parentId: issues.parentId, originKind: issues.originKind })
+        .from(issues)
+        .where(and(eq(issues.companyId, issue.companyId), eq(issues.id, parentId)))
+        .then((rows) => rows[0] ?? null);
+      if (!parent) return false;
+      if (parent.originKind === PRODUCTIVITY_REVIEW_ORIGIN_KIND) return true;
+      parentId = parent.parentId;
+      depth += 1;
+    }
+    return false;
+  }
+
+  async function findOpenProductivityReview(companyId: string, sourceIssueId: string) {
+    return db
+      .select()
+      .from(issues)
+      .where(
+        and(
+          eq(issues.companyId, companyId),
+          eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND),
+          eq(issues.originId, sourceIssueId),
+          isNull(issues.hiddenAt),
+          notInArray(issues.status, ["done", "cancelled"]),
+        ),
+      )
+      .orderBy(desc(issues.updatedAt))
+      .limit(1)
+      .then((rows) => rows[0] ?? null);
+  }
+
+  async function findRecentResolvedProductivityReview(
+    companyId: string,
+    sourceIssueId: string,
+    thresholds: ProductivityReviewThresholds,
+    now: Date,
+  ) {
+    const cutoff = new Date(now.getTime() - thresholds.resolvedSnoozeMs);
+    return db
+      .select({ id: issues.id, identifier: issues.identifier, status: issues.status, updatedAt: issues.updatedAt })
+      .from(issues)
+      .where(
+        and(
+          eq(issues.companyId, companyId),
+          eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND),
+          eq(issues.originId, sourceIssueId),
+          eq(issues.status, "done"),
+          gt(issues.updatedAt, cutoff),
+        ),
+      )
+      .orderBy(desc(issues.updatedAt))
+      .limit(1)
+      .then((rows) => rows[0] ?? null);
+  }
+
+  async function countIssueRunsSince(companyId: string, agentId: string, issueId: string, since: Date) {
+    return db
+      .select({ count: sql<number>`count(*)::int` })
+      .from(heartbeatRuns)
+      .where(
+        and(
+          eq(heartbeatRuns.companyId, companyId),
+          eq(heartbeatRuns.agentId, agentId),
+          issueRunScopeSql(issueId),
+          sql`coalesce(${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) >= ${since.toISOString()}::timestamptz`,
+        ),
+      )
+      .then((rows) => rows[0]?.count ?? 0);
+  }
+
+  async function countIssueCommentsSince(companyId: string, issueId: string, since?: Date) {
+    return db
+      .select({ count: sql<number>`count(*)::int` })
+      .from(issueComments)
+      .where(
+        and(
+          eq(issueComments.companyId, companyId),
+          eq(issueComments.issueId, issueId),
+          since ? sql`${issueComments.createdAt} >= ${since.toISOString()}::timestamptz` : undefined,
+        ),
+      )
+      .then((rows) => rows[0]?.count ?? 0);
+  }
+
+  async function collectEvidence(
+    sourceIssue: IssueRow,
+    sourceAgent: AgentRow,
+    thresholds: ProductivityReviewThresholds,
+    now: Date,
+  ): Promise<ProductivityReviewEvidence | null> {
+    const oneHourAgo = new Date(now.getTime() - 60 * 60 * 1000);
+    const sixHoursAgo = new Date(now.getTime() - 6 * 60 * 60 * 1000);
+
+    const latestRuns = await db
+      .select()
+      .from(heartbeatRuns)
+      .where(
+        and(
+          eq(heartbeatRuns.companyId, sourceIssue.companyId),
+          eq(heartbeatRuns.agentId, sourceAgent.id),
+          issueRunScopeSql(sourceIssue.id),
+        ),
+      )
+      .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
+      .limit(MAX_RUNS_FOR_STREAK);
+
+    const runIds = latestRuns.map((run) => run.id);
+    const commentRunIds = new Set<string>();
+    if (runIds.length > 0) {
+      const commentRows = await db
+        .select({ createdByRunId: issueComments.createdByRunId })
+        .from(issueComments)
+        .where(
+          and(
+            eq(issueComments.companyId, sourceIssue.companyId),
+            eq(issueComments.issueId, sourceIssue.id),
+            inArray(issueComments.createdByRunId, runIds),
+          ),
+        );
+      for (const row of commentRows) {
+        if (row.createdByRunId) commentRunIds.add(row.createdByRunId);
+      }
+    }
+
+    const terminalRuns = latestRuns.filter((run) =>
+      TERMINAL_RUN_STATUSES.includes(run.status as (typeof TERMINAL_RUN_STATUSES)[number]),
+    );
+    let noCommentStreak = 0;
+    for (const run of terminalRuns) {
+      if (commentRunIds.has(run.id)) break;
+      noCommentStreak += 1;
+    }
+
+    const [
+      runCountLastHour,
+      runCountLastSixHours,
+      commentCount,
+      commentCountLastHour,
+      commentCountLastSixHours,
+      latestComments,
+      costRow,
+    ] = await Promise.all([
+      countIssueRunsSince(sourceIssue.companyId, sourceAgent.id, sourceIssue.id, oneHourAgo),
+      countIssueRunsSince(sourceIssue.companyId, sourceAgent.id, sourceIssue.id, sixHoursAgo),
+      countIssueCommentsSince(sourceIssue.companyId, sourceIssue.id),
+      countIssueCommentsSince(sourceIssue.companyId, sourceIssue.id, oneHourAgo),
+      countIssueCommentsSince(sourceIssue.companyId, sourceIssue.id, sixHoursAgo),
+      db
+        .select()
+        .from(issueComments)
+        .where(and(eq(issueComments.companyId, sourceIssue.companyId), eq(issueComments.issueId, sourceIssue.id)))
+        .orderBy(desc(issueComments.createdAt), desc(issueComments.id))
+        .limit(5),
+      db
+        .select({ costCents: sql<number>`coalesce(sum(${costEvents.costCents}), 0)::int` })
+        .from(costEvents)
+        .where(and(eq(costEvents.companyId, sourceIssue.companyId), eq(costEvents.issueId, sourceIssue.id)))
+        .then((rows) => rows[0] ?? { costCents: 0 }),
+    ]);
+
+    const activeRunCount = latestRuns.filter((run) =>
+      ACTIVE_RUN_STATUSES.includes(run.status as (typeof ACTIVE_RUN_STATUSES)[number]),
+    ).length;
+    const activeStartedAt = sourceIssue.startedAt ?? sourceIssue.executionLockedAt ?? null;
+    const elapsedMs = sourceIssue.status === "in_progress" && activeStartedAt
+      ? Math.max(0, now.getTime() - activeStartedAt.getTime())
+      : null;
+
+    const noComment = noCommentStreak >= thresholds.noCommentStreakRuns;
+    const longActive = elapsedMs !== null && elapsedMs >= thresholds.longActiveMs;
+    const highChurn =
+      runCountLastHour >= thresholds.highChurnHourly ||
+      commentCountLastHour >= thresholds.highChurnHourly ||
+      runCountLastSixHours >= thresholds.highChurnSixHours ||
+      commentCountLastSixHours >= thresholds.highChurnSixHours;
+    const trigger = choosePrimaryTrigger({ noComment, longActive, highChurn });
+    if (!trigger) return null;
+
+    const triggerReasons: string[] = [];
+    if (noComment) triggerReasons.push(`${noCommentStreak} consecutive completed issue-linked runs had no run-created issue comment`);
+    if (longActive) triggerReasons.push(`current active episode has lasted ${msToHuman(elapsedMs)}`);
+    if (highChurn) {
+      triggerReasons.push(
+        `${runCountLastHour} runs/${commentCountLastHour} comments in 1h; ${runCountLastSixHours} runs/${commentCountLastSixHours} comments in 6h`,
+      );
+    }
+
+    return {
+      trigger,
+      triggerReasons,
+      sourceIssue,
+      sourceAgent,
+      noCommentStreak,
+      totalRunCount: latestRuns.length,
+      terminalRunCount: terminalRuns.length,
+      activeRunCount,
+      runCountLastHour,
+      runCountLastSixHours,
+      commentCount,
+      commentCountLastHour,
+      commentCountLastSixHours,
+      elapsedMs,
+      latestRuns: latestRuns.slice(0, 5),
+      latestComments,
+      costCents: costRow.costCents,
+      usageSamples: latestRuns
+        .filter((run) => run.usageJson)
+        .slice(0, 3)
+        .map((run) => ({ runId: run.id, usageJson: run.usageJson ?? null })),
+      nextAction: latestRuns.find((run) => run.nextAction)?.nextAction ?? null,
+      thresholds,
+      generatedAt: now,
+    };
+  }
+
+  async function resolveReviewOwnerAgentId(sourceIssue: IssueRow, sourceAgent: AgentRow) {
+    const candidateIds: string[] = [];
+    if (sourceAgent.reportsTo) candidateIds.push(sourceAgent.reportsTo);
+    if (sourceIssue.createdByAgentId) candidateIds.push(sourceIssue.createdByAgentId);
+    if (sourceIssue.projectId) {
+      const project = await db
+        .select({ leadAgentId: projects.leadAgentId })
+        .from(projects)
+        .where(and(eq(projects.companyId, sourceIssue.companyId), eq(projects.id, sourceIssue.projectId)))
+        .then((rows) => rows[0] ?? null);
+      if (project?.leadAgentId) candidateIds.push(project.leadAgentId);
+    }
+    const roleCandidates = await db
+      .select({ id: agents.id })
+      .from(agents)
+      .where(and(eq(agents.companyId, sourceIssue.companyId), inArray(agents.role, ["cto", "ceo"])))
+      .orderBy(sql`case when ${agents.role} = 'cto' then 0 else 1 end`, asc(agents.createdAt), asc(agents.id));
+    candidateIds.push(...roleCandidates.map((agent) => agent.id));
+
+    const seen = new Set<string>();
+    for (const agentId of candidateIds) {
+      if (seen.has(agentId)) continue;
+      seen.add(agentId);
+      const candidate = await getAgent(agentId);
+      if (!candidate || candidate.companyId !== sourceIssue.companyId || !isAgentInvokable(candidate)) continue;
+      const budgetBlock = await budgets.getInvocationBlock(sourceIssue.companyId, candidate.id, {
+        issueId: sourceIssue.id,
+        projectId: sourceIssue.projectId ?? null,
+      });
+      if (!budgetBlock) return candidate.id;
+    }
+    return null;
+  }
+
+  function buildReviewMarkdown(evidence: ProductivityReviewEvidence, prefix: string) {
+    const latestRuns = evidence.latestRuns.length > 0
+      ? evidence.latestRuns.map((run) =>
+        `- ${runUiLink(run, prefix)} \`${run.status}\` liveness \`${run.livenessState ?? "unknown"}\`, created ${run.createdAt.toISOString()}${run.nextAction ? `, next action: ${truncateInline(run.nextAction, 160)}` : ""}`,
+      ).join("\n")
+      : "- none";
+    const latestComments = evidence.latestComments.length > 0
+      ? evidence.latestComments.map((comment) =>
+        `- ${comment.createdAt.toISOString()}${comment.createdByRunId ? ` run \`${comment.createdByRunId}\`` : ""}: ${truncateInline(comment.body)}`,
+      ).join("\n")
+      : "- none";
+    const usage = evidence.usageSamples.length > 0
+      ? evidence.usageSamples.map((sample) => `- \`${sample.runId}\`: \`${JSON.stringify(sample.usageJson).slice(0, 500)}\``).join("\n")
+      : "- no usage payloads on sampled runs";
+    return [
+      "Paperclip detected an unusual productivity/progression pattern on an assigned issue.",
+      "",
+      "## Source",
+      "",
+      `- Source issue: ${issueUiLink(evidence.sourceIssue, prefix)}`,
+      `- Assigned agent: ${evidence.sourceAgent.name} (${evidence.sourceAgent.role})`,
+      `- Primary trigger: \`${evidence.trigger}\` (${formatTrigger(evidence.trigger)})`,
+      `- Trigger reasons: ${evidence.triggerReasons.join("; ")}`,
+      `- Generated at: ${evidence.generatedAt.toISOString()}`,
+      "",
+      "## Evidence",
+      "",
+      `- Total sampled issue-linked runs: ${evidence.totalRunCount}`,
+      `- Terminal sampled runs: ${evidence.terminalRunCount}`,
+      `- Active queued/running/scheduled runs: ${evidence.activeRunCount}`,
+      `- No-comment completed-run streak: ${evidence.noCommentStreak}`,
+      `- Current active elapsed time: ${msToHuman(evidence.elapsedMs)}`,
+      `- Runs in rolling windows: ${evidence.runCountLastHour}/1h, ${evidence.runCountLastSixHours}/6h`,
+      `- Comments total/window: ${evidence.commentCount} total, ${evidence.commentCountLastHour}/1h, ${evidence.commentCountLastSixHours}/6h`,
+      `- Cost events total: ${evidence.costCents} cents`,
+      `- Current next action: ${evidence.nextAction ? truncateInline(evidence.nextAction, 500) : "none recorded"}`,
+      "",
+      "## Thresholds",
+      "",
+      `- No-comment streak: ${evidence.thresholds.noCommentStreakRuns} completed runs`,
+      `- Long active duration: ${msToHuman(evidence.thresholds.longActiveMs)}`,
+      `- High churn: ${evidence.thresholds.highChurnHourly}/1h or ${evidence.thresholds.highChurnSixHours}/6h runs/comments`,
+      `- Resolved-review snooze: ${msToHuman(evidence.thresholds.resolvedSnoozeMs)}`,
+      "",
+      "## Latest Runs",
+      "",
+      latestRuns,
+      "",
+      "## Latest Comments",
+      "",
+      latestComments,
+      "",
+      "## Usage Samples",
+      "",
+      usage,
+      "",
+      "## Manager Decision",
+      "",
+      "- Close as productive if this pattern is expected.",
+      "- Continue with a snooze window if the current work should keep running without repeat review spam.",
+      "- Request decomposition, reroute, block with an unblock owner, or stop/cancel the source work if the work is inefficient.",
+    ].join("\n");
+  }
+
+  function buildRefreshComment(evidence: ProductivityReviewEvidence, prefix: string) {
+    return [
+      "Productivity review evidence refreshed.",
+      "",
+      `- Source issue: ${issueUiLink(evidence.sourceIssue, prefix)}`,
+      `- Trigger: \`${evidence.trigger}\` (${formatTrigger(evidence.trigger)})`,
+      `- Reasons: ${evidence.triggerReasons.join("; ")}`,
+      `- No-comment streak: ${evidence.noCommentStreak}`,
+      `- Runs/comments: ${evidence.runCountLastHour}/${evidence.commentCountLastHour} in 1h, ${evidence.runCountLastSixHours}/${evidence.commentCountLastSixHours} in 6h`,
+      `- Next action: ${evidence.nextAction ? truncateInline(evidence.nextAction, 300) : "none recorded"}`,
+    ].join("\n");
+  }
+
+  async function createOrUpdateReview(
+    evidence: ProductivityReviewEvidence,
+    opts: { prefix: string },
+  ) {
+    const existing = await findOpenProductivityReview(evidence.sourceIssue.companyId, evidence.sourceIssue.id);
+    if (existing) {
+      await issuesSvc.addComment(existing.id, buildRefreshComment(evidence, opts.prefix), {});
+      await logActivity(db, {
+        companyId: evidence.sourceIssue.companyId,
+        actorType: "system",
+        actorId: "system",
+        action: "issue.productivity_review_updated",
+        entityType: "issue",
+        entityId: existing.id,
+        agentId: existing.assigneeAgentId,
+        details: {
+          source: "productivity_review.reconcile",
+          sourceIssueId: evidence.sourceIssue.id,
+          trigger: evidence.trigger,
+          noCommentStreak: evidence.noCommentStreak,
+          runCountLastHour: evidence.runCountLastHour,
+          commentCountLastHour: evidence.commentCountLastHour,
+        },
+      });
+      return { kind: "updated" as const, reviewIssueId: existing.id };
+    }
+
+    const ownerAgentId = await resolveReviewOwnerAgentId(evidence.sourceIssue, evidence.sourceAgent);
+    let review: Awaited<ReturnType<typeof issuesSvc.create>>;
+    try {
+      review = await issuesSvc.create(evidence.sourceIssue.companyId, {
+        title: `Review productivity for ${evidence.sourceIssue.identifier ?? evidence.sourceIssue.title}`,
+        description: buildReviewMarkdown(evidence, opts.prefix),
+        status: "todo",
+        priority: evidence.trigger === "long_active_duration" ? "medium" : "high",
+        parentId: evidence.sourceIssue.id,
+        projectId: evidence.sourceIssue.projectId,
+        goalId: evidence.sourceIssue.goalId,
+        billingCode: evidence.sourceIssue.billingCode,
+        assigneeAgentId: ownerAgentId,
+        originKind: PRODUCTIVITY_REVIEW_ORIGIN_KIND,
+        originId: evidence.sourceIssue.id,
+        originFingerprint: productivityReviewFingerprint(evidence.sourceIssue.id),
+        requestDepth: evidence.sourceIssue.requestDepth + 1,
+      });
+    } catch (error) {
+      const maybe = error as { code?: string; constraint?: string; message?: string };
+      const uniqueConflict = maybe.code === "23505" &&
+        (
+          maybe.constraint === "issues_active_productivity_review_uq" ||
+          typeof maybe.message === "string" && maybe.message.includes("issues_active_productivity_review_uq")
+        );
+      if (!uniqueConflict) throw error;
+      const raced = await findOpenProductivityReview(evidence.sourceIssue.companyId, evidence.sourceIssue.id);
+      if (!raced) throw error;
+      return { kind: "existing" as const, reviewIssueId: raced.id };
+    }
+
+    await logActivity(db, {
+      companyId: evidence.sourceIssue.companyId,
+      actorType: "system",
+      actorId: "system",
+      action: "issue.productivity_review_created",
+      entityType: "issue",
+      entityId: review.id,
+      agentId: ownerAgentId,
+      details: {
+        source: "productivity_review.reconcile",
+        sourceIssueId: evidence.sourceIssue.id,
+        trigger: evidence.trigger,
+        noCommentStreak: evidence.noCommentStreak,
+        runCountLastHour: evidence.runCountLastHour,
+        commentCountLastHour: evidence.commentCountLastHour,
+      },
+    });
+
+    if (ownerAgentId && deps?.enqueueWakeup) {
+      await deps.enqueueWakeup(ownerAgentId, {
+        source: "assignment",
+        triggerDetail: "system",
+        reason: "issue_assigned",
+        payload: {
+          issueId: review.id,
+          sourceIssueId: evidence.sourceIssue.id,
+          trigger: evidence.trigger,
+        },
+        requestedByActorType: "system",
+        requestedByActorId: "productivity_review",
+        contextSnapshot: {
+          issueId: review.id,
+          taskId: review.id,
+          wakeReason: "issue_assigned",
+          source: PRODUCTIVITY_REVIEW_ORIGIN_KIND,
+          sourceIssueId: evidence.sourceIssue.id,
+          productivityReviewTrigger: evidence.trigger,
+        },
+      });
+    }
+
+    return { kind: "created" as const, reviewIssueId: review.id };
+  }
+
+  async function reconcileProductivityReviews(opts?: {
+    now?: Date;
+    companyId?: string;
+    thresholds?: Partial<ProductivityReviewThresholds>;
+  }) {
+    const now = opts?.now ?? new Date();
+    const thresholds = buildThresholds(opts?.thresholds);
+    const candidates = await db
+      .select()
+      .from(issues)
+      .where(
+        and(
+          opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
+          isNull(issues.hiddenAt),
+          isNull(issues.assigneeUserId),
+          inArray(issues.status, ["todo", "in_progress"]),
+          sql`${issues.assigneeAgentId} is not null`,
+          sql`${issues.originKind} <> ${PRODUCTIVITY_REVIEW_ORIGIN_KIND}`,
+        ),
+      )
+      .orderBy(asc(issues.updatedAt), asc(issues.id))
+      .limit(MAX_CANDIDATE_ISSUES);
+
+    const result = {
+      scanned: candidates.length,
+      created: 0,
+      updated: 0,
+      existing: 0,
+      snoozed: 0,
+      skipped: 0,
+      reviewIssueIds: [] as string[],
+    };
+
+    const prefixCache = new Map<string, string>();
+    for (const candidate of candidates) {
+      if (!candidate.assigneeAgentId) {
+        result.skipped += 1;
+        continue;
+      }
+      if (await isProductivityReviewDescendant(candidate)) {
+        result.skipped += 1;
+        continue;
+      }
+      if (await findRecentResolvedProductivityReview(candidate.companyId, candidate.id, thresholds, now)) {
+        result.snoozed += 1;
+        continue;
+      }
+      const sourceAgent = await getAgent(candidate.assigneeAgentId);
+      if (!sourceAgent || sourceAgent.companyId !== candidate.companyId) {
+        result.skipped += 1;
+        continue;
+      }
+      const evidence = await collectEvidence(candidate, sourceAgent, thresholds, now);
+      if (!evidence) {
+        result.skipped += 1;
+        continue;
+      }
+      let prefix = prefixCache.get(candidate.companyId);
+      if (!prefix) {
+        prefix = await getCompanyIssuePrefix(candidate.companyId);
+        prefixCache.set(candidate.companyId, prefix);
+      }
+      const outcome = await createOrUpdateReview(evidence, { prefix });
+      if (outcome.kind === "created") result.created += 1;
+      else if (outcome.kind === "updated") result.updated += 1;
+      else result.existing += 1;
+      result.reviewIssueIds.push(outcome.reviewIssueId);
+    }
+
+    return result;
+  }
+
+  async function isProductivityReviewContinuationHoldActive(input: {
+    companyId: string;
+    issueId: string;
+    agentId: string;
+    now?: Date;
+    thresholds?: Partial<ProductivityReviewThresholds>;
+  }) {
+    const now = input.now ?? new Date();
+    const thresholds = buildThresholds(input.thresholds);
+    const [sourceIssue, sourceAgent, openReview] = await Promise.all([
+      db
+        .select()
+        .from(issues)
+        .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId)))
+        .then((rows) => rows[0] ?? null),
+      getAgent(input.agentId),
+      findOpenProductivityReview(input.companyId, input.issueId),
+    ]);
+    if (!sourceIssue || !sourceAgent || !openReview) return { held: false as const };
+    if (sourceAgent.companyId !== input.companyId) return { held: false as const };
+    const evidence = await collectEvidence(sourceIssue, sourceAgent, thresholds, now);
+    if (!evidence || !isSoftStopTrigger(evidence.trigger)) return { held: false as const };
+    return {
+      held: true as const,
+      reviewIssueId: openReview.id,
+      reviewIdentifier: openReview.identifier,
+      trigger: evidence.trigger,
+      reason: evidence.triggerReasons.join("; "),
+    };
+  }
+
+  async function recordContinuationHold(input: {
+    companyId: string;
+    issueId: string;
+    runId: string;
+    agentId: string;
+    reviewIssueId: string;
+    trigger: ProductivityReviewTrigger;
+    reason: string;
+  }) {
+    await logActivity(db, {
+      companyId: input.companyId,
+      actorType: "system",
+      actorId: "system",
+      agentId: input.agentId,
+      runId: input.runId,
+      action: "issue.productivity_review_continuation_held",
+      entityType: "issue",
+      entityId: input.issueId,
+      details: {
+        source: "productivity_review.continuation_hold",
+        reviewIssueId: input.reviewIssueId,
+        trigger: input.trigger,
+        reason: input.reason,
+      },
+    });
+  }
+
+  return {
+    reconcileProductivityReviews,
+    isProductivityReviewContinuationHoldActive,
+    recordContinuationHold,
+  };
+}

server/src/services/recovery/origins.ts

@@ -1,5 +1,6 @@
 export const RECOVERY_ORIGIN_KINDS = {
   issueGraphLivenessEscalation: "harness_liveness_escalation",
+  issueProductivityReview: "issue_productivity_review",
   strandedIssueRecovery: "stranded_issue_recovery",
   staleActiveRunEvaluation: "stale_active_run_evaluation",
 } as const;