fix(server): preserve native progress cursor precision and isolate projection

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

doc/plans/2026-10-25-native-chat-progress-sleep-audit.md

@@ -0,0 +1,142 @@
+# Native chat progress: sleep migration audit
+
+Parent: #17606 at `ed76256b979541b962fa784016a550c32920d670`.
+Branch: `codex/sleep-29-native-chat-progress`.
+Status: groundwork verified; production still polls. Do not mark migrated.
+
+## Duty and owner
+
+Native progress is the separate `projectNativeProgress` lane in the one-second
+chat reconciliation coordinator. It projects a closed allowlist of durable run
+event types into coarse provider-safe phrases. It never copies event messages,
+payloads, tool names, arguments, or results into external publications.
+
+`chat-native-progress.ts` owns selection, authority checks and insertion.
+`safe-native-chat-progress.ts` owns the normalized allowlist and phrase policy.
+`chat-run-publication-policy.ts` holds only the existing projection window and
+shared automatic-publication eligibility. The latter preserves the distinction
+between upgraded GitHub bots (tool-owned replies) and legacy automatic replies.
+The original export and explicit combined lifecycle/progress API stay compatible.
+No generic scheduler should interpret provider payloads or decide disclosure.
+
+## Existing durable contract
+
+- Native running run, matching company/agent and bound issue.
+- Active/waiting conversation with an automatic endpoint assigned to that agent.
+- Existing base working publication in pending/retry/streaming/published state.
+- Latest allowlisted event for that run, inside the existing 24-hour window.
+- Event time at least 20 seconds after the base working publication.
+- Exact current origin/lineage authorization; current provider interaction and
+  explicitly authored final output may suppress this optional progress.
+- At most one publication per phase. Twenty-second spacing compares the saved
+  event time with saved publication times. Passage of wall-clock time does not
+  make the same suppressed old event eligible. Do not invent a timer to retry it
+  after twenty seconds or change the cadence policy as part of this migration.
+- Existing publications provide durable deduplication and outbound delivery.
+  Do not introduce another queue table.
+
+## Verified pagination repair
+
+The selector previously round-tripped event timestamps through JavaScript Date.
+PostgreSQL microseconds were lost. With a full page of rejected candidates at
+one fractional timestamp, every next page selected those same candidates; an
+eligible event later in the ordered stream was never reached.
+
+The new regression creates 30 origin-rejected candidates with a microsecond
+timestamp and a valid event after them. A query-count guard fails the broken
+pagination without leaving an infinite SQL chain after the test timeout.
+The fix keeps the timestamp as PostgreSQL text and uses the full timestamp/run/
+conversation tuple for continuation, preserving ordering and eligibility.
+
+- `/tmp/sleep29-progress-cursor-before.log`: fails with non-advancing pagination.
+- `/tmp/sleep29-progress-cursor-acceptance.log`: 17 selected cases pass after repair.
+- `/tmp/sleep29-progress-owner-acceptance.log`: 21 native-progress and combined
+  projection cases pass after extracting the owner.
+- `/tmp/sleep29-progress-owner-typecheck.log`: server production typecheck passes.
+  This excludes test files. No worker-quiet or production migration claim yet.
+- Database: isolated external PostgreSQL
+  `paperclip_sleep28_review_chat_20261025_test`, through
+  `PAPERCLIP_TEST_DATABASE_URL`. Tests retire their publication endpoints.
+
+## Producer audit and remaining implementation
+
+Two event persistence owners need coverage, not just the post-commit live event:
+
+1. `heartbeat-run-events.ts::appendHeartbeatRunEvent`: native ingestion uses its
+   transaction/deduplication path. `paperclip-control-plane-port.ts` wraps this
+   in an outer transaction with active-authority checks before and after append.
+   Intent must precede the event write and follow outer commit/rollback. Duplicate
+   replay must not create fresh projection work. Ordinary non-progress log events
+   must not acquire a transaction/XID merely for this feature.
+2. `native-run-coordinator-store.ts::appendEvent`: directly inserts run events
+   after current-run and sequence/duplicate checks. It does not call the generic
+   append helper. Cover it explicitly in its existing transaction.
+
+Additional paths found:
+
+- `provider-trace-workspace-diff-reprojection.ts` appends
+  `workspace.diff.recorded` through the generic native-source path. Recovery can
+  therefore create a new eligible event independently of live runner ingress.
+- `heartbeat/run-lifecycle.ts` uses the generic writer for dynamic event types.
+  Preserve its redaction/runtime-status responsibilities; do not project raw data.
+- Direct insertion in `heartbeat/queue.ts` writes a lifecycle event; it is outside
+  the closed progress allowlist. Coordinator-store direct insertion is not.
+- Other native helper callers currently append lifecycle, process, cleanup, or
+  credential events. Their existence is not permission to signal every log entry.
+
+Eligibility changes can make a saved event actionable without a new event:
+run/issue binding and state, conversation state, current identity/membership,
+endpoint/connection/resource authority, lineage receipts, interaction arbitration,
+and base-publication state. Reuse the existing issue/company-scoped notification
+boundary audited in slice 28. Extend that shared boundary once; avoid repeated
+recovery wrappers across routes. Inspect all base-publication retry and operator
+resolution owners before removing the poll. `notifyChatPublicationWork` already
+accepts an optional issue scope for projection-affecting changes; transport claim
+bookkeeping intentionally does not invalidate projection.
+
+## Contention and scheduling
+
+The current projector locks issue then run with SKIP LOCKED. This order avoids
+inversion with native finalization and publication foreign keys. A zero insert
+count currently conflates ordinary suppression, deletion/terminalization, and
+lock contention. The polling caller retries all of them later.
+
+Before replacing that caller, expose whether concrete candidates were deferred
+by contention. Retain a retry only for unresolved candidates or failure. Recheck
+existence/current eligibility so a deleted or terminal run does not become an
+endless retry. A later committed event or eligibility change must remain dirty
+if it arrives during the current sweep.
+
+Use the shared scoped-deadline worker and delivery coordinator for coalescing,
+company fairness, failure backoff, drain/resume and shutdown. Reconstruct relevant
+issue scopes at startup. Scope refresh and projection must use the same candidate
+contract. A short settled pass disarms; unfinished batches retain work. Preserve
+24-hour policy rather than broadening history during this migration.
+
+## Required acceptance before removing periodic dispatch
+
+- Configured idle and retained completed/suppressed history generate no recurring
+  worker SQL after startup and settlement.
+- Real event producers wake after outer commit, not savepoint release; rollback
+  publishes nothing and settles uncertainty through the existing protocol.
+- Duplicate native-event replay and irrelevant log entries create no new debt.
+- Both event persistence owners and trace reprojection have causal coverage.
+- Restored authority/base-publication state wakes saved eligible events.
+- Contended issue and run locks retain recovery while unrelated work proceeds;
+  terminalization, final answers and changed authority win before retry.
+- Full batches, submillisecond cursor progression, cadence and phase deduplication.
+- Restart, failure, changes during a pass, drain/resume, and shutdown.
+- Assembled startup registers the worker and removes this one periodic lane;
+  other chat maintenance lanes remain until their own migrations.
+- Current-head checks/review, final base reconciliation, and an accurate PR body.
+
+Process-local signals retain the single active writer model. No Cloud sleep gate
+or external alarm coverage changes here; worker quiet is not provider suspension.
+
+## Remote reconciliation note
+
+Fetched master is `e302f2098f` (not merged into this stack yet). Newly landed
+#17616 expands native-harness backup expiration to continuity/quarantine evidence
+and adds its evidence signal. Its index.ts interval still exists. Audit this newer
+contract when migrating that separate retention owner; do not recreate or drop
+its preservation guarantees based on the old inventory.

server/src/__tests__/chat-channels.integration.test.ts

@@ -63567,6 +63567,52 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     };
   }
 
+  it("advances native progress past suppressed candidates with microsecond timestamps", async () => {
+    const context = await safeNativeProgressFixture("telegram", "119");
+    try {
+      const valid = await context.createRun("eligible after rejected page");
+      const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, valid.runId));
+      const rejectedIds = Array.from({ length: 30 }, () => randomUUID());
+      await db.insert(heartbeatRuns).values(rejectedIds.map(id => ({
+        id, companyId: source.companyId, agentId: source.agentId,
+        runtimeMode: "native", status: "running",
+        contextSnapshot: { ...source.contextSnapshot, endpointId: randomUUID() },
+      })));
+      await db.insert(chatPublications).values(rejectedIds.map(id => ({
+        companyId: context.fixture.companyId, endpointId: context.endpoint.id,
+        conversationId: context.conversation.id, issueId: context.conversation.issueId,
+        idempotencyKey: `run:${id}:working:${context.endpoint.id}`,
+        payload: { text: "Maya is working…", progressState: "working" },
+        state: "published", createdAt: valid.baseCreatedAt,
+      })));
+      const eventAt = new Date(valid.baseCreatedAt.getTime() + 30_000);
+      await db.insert(heartbeatRunEvents).values(rejectedIds.map(runId => ({
+        companyId: source.companyId, runId, agentId: source.agentId,
+        seq: 1, eventType: "item.completed",
+        createdAt: sql`${eventAt.toISOString()}::timestamptz + interval '0.000123 seconds'`,
+      })));
+      await context.addEvent(valid, "item.completed", 1, new Date(eventAt.getTime() + 1));
+      const select = db.select.bind(db);
+      let pages = 0;
+      const guard = vi.spyOn(db, "select").mockImplementation((...args) => {
+        if (args[0] && "eventCreatedAt" in args[0] && ++pages > 10) {
+          throw new Error("Native progress pagination did not advance past suppressed candidates");
+        }
+        return select(...args);
+      });
+      try {
+        await expect(enqueueSafeNativeChatProgress(db, { since: eventAt, limit: 1 })).resolves.toBe(1);
+      } finally { guard.mockRestore(); }
+      const progress = await db.select().from(chatPublications).where(and(
+        eq(chatPublications.conversationId, context.conversation.id),
+        like(chatPublications.idempotencyKey, "%:native:%"),
+      ));
+      expect(progress).toEqual([expect.objectContaining({
+        idempotencyKey: `run:${valid.runId}:working:${context.endpoint.id}:native:making_progress:1`,
+      })]);
+    } finally { await retirePublicationFixture(context.service, context.endpoint.id); }
+  });
+
   it.each(["separate owners", "combined budget"] as const)(
     "keeps lifecycle and native progress projection independent with %s",
     async (mode) => {

server/src/services/chat-native-progress.ts

@@ -0,0 +1,393 @@
+import { and, asc, desc, eq, gt, gte, inArray, like, notExists, sql } from "drizzle-orm";
+import { alias } from "drizzle-orm/pg-core";
+import type { Db } from "@paperclipai/db";
+import { agents, chatConversations, chatEndpoints, chatPublications, heartbeatRunEvents, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
+import { projectSafeChatPublication } from "./chat-publication-projection.js";
+import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js";
+import { resolveChatOriginPublicationBindings } from "./issues.js";
+import { boundRunIssueIdSql } from "./run-task-binding.js";
+import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js";
+import { SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, safeNativeChatProgressForEvent } from "./safe-native-chat-progress.js";
+
+type SafeNativeChatProgressCandidate = {
+  eventId: number;
+  eventSeq: number;
+  eventType: string;
+  eventCreatedAt: string;
+  runId: string;
+  agentId: string;
+  issueId: string;
+  companyId: string;
+  endpointId: string;
+  conversationId: string;
+  agentName: string;
+};
+
+const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000;
+
+/** Only closed event types become coarse progress; event payloads stay local. */
+export async function enqueueSafeNativeChatProgress(
+  db: Db,
+  options: { since?: Date; limit?: number } = {},
+): Promise<number> {
+  const input = chatRunProjectionWindow(options);
+  const basePublication = alias(
+    chatPublications,
+    "safe_native_progress_base_publication",
+  );
+  const laterEvent = alias(
+    heartbeatRunEvents,
+    "safe_native_progress_later_event",
+  );
+  const issueIdFromContext = boundRunIssueIdSql();
+  let inserted = 0;
+  let cursor: {
+    eventCreatedAt: string;
+    runId: string;
+    conversationId: string;
+  } | null = null;
+
+  while (inserted < input.limit) {
+    const pageSize = Math.max(25, Math.min(200, input.limit - inserted));
+    const pageCursor: typeof cursor = cursor;
+    const rows: SafeNativeChatProgressCandidate[] = await db
+      .select({
+        eventId: heartbeatRunEvents.id,
+        eventSeq: heartbeatRunEvents.seq,
+        eventType: heartbeatRunEvents.eventType,
+        // Keep PostgreSQL microseconds in the cursor. Date truncation can
+        // repeatedly select the same suppressed page without advancing.
+        eventCreatedAt: sql<string>`${heartbeatRunEvents.createdAt}::text`,
+        runId: heartbeatRuns.id,
+        agentId: heartbeatRuns.agentId,
+        issueId: chatConversations.issueId,
+        companyId: chatConversations.companyId,
+        endpointId: chatConversations.endpointId,
+        conversationId: chatConversations.id,
+        agentName: agents.name,
+      })
+      .from(heartbeatRunEvents)
+      .innerJoin(
+        heartbeatRuns,
+        and(
+          eq(heartbeatRuns.id, heartbeatRunEvents.runId),
+          eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId),
+          eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId),
+          eq(heartbeatRuns.runtimeMode, "native"),
+          eq(heartbeatRuns.status, "running"),
+        ),
+      )
+      .innerJoin(
+        chatConversations,
+        and(
+          eq(chatConversations.companyId, heartbeatRuns.companyId),
+          sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`,
+          inArray(chatConversations.state, ["active", "waiting"]),
+        ),
+      )
+      .innerJoin(
+        chatEndpoints,
+        and(
+          eq(chatEndpoints.companyId, chatConversations.companyId),
+          eq(chatEndpoints.id, chatConversations.endpointId),
+          eq(chatEndpoints.publicationMode, "automatic"),
+          allowsAutomaticRunPublication(),
+          eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId),
+        ),
+      )
+      .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
+      .innerJoin(
+        basePublication,
+        and(
+          eq(basePublication.companyId, heartbeatRuns.companyId),
+          eq(basePublication.endpointId, chatConversations.endpointId),
+          eq(basePublication.conversationId, chatConversations.id),
+          eq(basePublication.issueId, chatConversations.issueId),
+          sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`,
+          inArray(basePublication.state, [
+            "pending",
+            "retry",
+            "streaming",
+            "published",
+          ]),
+        ),
+      )
+      .where(
+        and(
+          inArray(heartbeatRunEvents.eventType, [
+            ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
+          ]),
+          gte(heartbeatRunEvents.createdAt, input.since),
+          sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`,
+          notExists(
+            db
+              .select({ id: laterEvent.id })
+              .from(laterEvent)
+              .where(
+                and(
+                  eq(laterEvent.companyId, heartbeatRunEvents.companyId),
+                  eq(laterEvent.runId, heartbeatRunEvents.runId),
+                  gt(laterEvent.seq, heartbeatRunEvents.seq),
+                  inArray(laterEvent.eventType, [
+                    ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
+                  ]),
+                ),
+              ),
+          ),
+          pageCursor
+            ? sql`(${heartbeatRunEvents.createdAt}, ${heartbeatRuns.id}, ${chatConversations.id}) >
+                (${pageCursor.eventCreatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)`
+            : undefined,
+        ),
+      )
+      .orderBy(
+        asc(heartbeatRunEvents.createdAt),
+        asc(heartbeatRuns.id),
+        asc(chatConversations.id),
+      )
+      .limit(pageSize);
+    if (rows.length === 0) break;
+    const lastRow = rows.at(-1)!;
+    cursor = {
+      eventCreatedAt: lastRow.eventCreatedAt,
+      runId: lastRow.runId,
+      conversationId: lastRow.conversationId,
+    };
+
+    for (const row of rows) {
+      if (inserted >= input.limit) break;
+      const result = await db.transaction(async (tx) => {
+        // Keep the normal issue -> run order used by native finalization. The
+        // publication insert takes an issue FK lock, so locking the run first
+        // would invert that order against a concurrent terminal commit. A
+        // contended issue/run is optional progress, not a reason to hold other
+        // chats: skip it now and revisit it with a fresh cursor next sweep.
+        const currentIssue = await tx
+          .select({ id: issues.id })
+          .from(issues)
+          .where(
+            and(
+              eq(issues.id, row.issueId),
+              eq(issues.companyId, row.companyId),
+            ),
+          )
+          .for("update", { skipLocked: true })
+          .then((currentRows) => currentRows[0] ?? null);
+        if (!currentIssue) return 0;
+        // Native event admission serializes on the run row. Taking the same
+        // short lock makes terminal status, a later event, or a final comment
+        // win before this non-terminal projection can be inserted.
+        const currentRun = await tx
+          .select({ id: heartbeatRuns.id })
+          .from(heartbeatRuns)
+          .where(
+            and(
+              eq(heartbeatRuns.id, row.runId),
+              eq(heartbeatRuns.companyId, row.companyId),
+              eq(heartbeatRuns.agentId, row.agentId),
+              eq(heartbeatRuns.runtimeMode, "native"),
+              eq(heartbeatRuns.status, "running"),
+            ),
+          )
+          .for("update", { skipLocked: true })
+          .then((currentRows) => currentRows[0] ?? null);
+        if (!currentRun) return 0;
+
+        const destination = await tx
+          .select({ id: chatConversations.id })
+          .from(chatConversations)
+          .innerJoin(
+            chatEndpoints,
+            and(
+              eq(chatEndpoints.companyId, chatConversations.companyId),
+              eq(chatEndpoints.id, chatConversations.endpointId),
+              eq(chatEndpoints.publicationMode, "automatic"),
+              allowsAutomaticRunPublication(),
+              eq(chatEndpoints.assignedAgentId, row.agentId),
+            ),
+          )
+          .where(
+            and(
+              eq(chatConversations.companyId, row.companyId),
+              eq(chatConversations.id, row.conversationId),
+              eq(chatConversations.endpointId, row.endpointId),
+              eq(chatConversations.issueId, row.issueId),
+              inArray(chatConversations.state, ["active", "waiting"]),
+            ),
+          )
+          .limit(1)
+          .then((currentRows) => currentRows[0] ?? null);
+        if (!destination) return 0;
+
+        const [currentEvent] = await tx
+          .select({
+            id: heartbeatRunEvents.id,
+            seq: heartbeatRunEvents.seq,
+            eventType: heartbeatRunEvents.eventType,
+            createdAt: heartbeatRunEvents.createdAt,
+          })
+          .from(heartbeatRunEvents)
+          .where(
+            and(
+              eq(heartbeatRunEvents.companyId, row.companyId),
+              eq(heartbeatRunEvents.runId, row.runId),
+              inArray(heartbeatRunEvents.eventType, [
+                ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
+              ]),
+            ),
+          )
+          .orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id))
+          .limit(1);
+        if (
+          !currentEvent ||
+          currentEvent.id !== row.eventId ||
+          currentEvent.seq !== row.eventSeq ||
+          currentEvent.eventType !== row.eventType
+        ) {
+          return 0;
+        }
+        const progress = safeNativeChatProgressForEvent(
+          currentEvent.eventType,
+          row.agentName,
+        );
+        if (!progress) return 0;
+
+        const baseKey = `run:${row.runId}:working:${row.endpointId}`;
+        const [currentBase] = await tx
+          .select({ createdAt: chatPublications.createdAt })
+          .from(chatPublications)
+          .where(
+            and(
+              eq(chatPublications.companyId, row.companyId),
+              eq(chatPublications.endpointId, row.endpointId),
+              eq(chatPublications.conversationId, row.conversationId),
+              eq(chatPublications.issueId, row.issueId),
+              eq(chatPublications.idempotencyKey, baseKey),
+              inArray(chatPublications.state, [
+                "pending",
+                "retry",
+                "streaming",
+                "published",
+              ]),
+            ),
+          )
+          .limit(1);
+        if (!currentBase) return 0;
+
+        const progressKeyPrefix = `${baseKey}:native:`;
+        const previousProgress = await tx
+          .select({
+            createdAt: chatPublications.createdAt,
+            idempotencyKey: chatPublications.idempotencyKey,
+          })
+          .from(chatPublications)
+          .where(
+            and(
+              eq(chatPublications.companyId, row.companyId),
+              eq(chatPublications.endpointId, row.endpointId),
+              eq(chatPublications.conversationId, row.conversationId),
+              eq(chatPublications.issueId, row.issueId),
+              like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`),
+            ),
+          );
+        if (
+          previousProgress.some((publication) =>
+            publication.idempotencyKey.startsWith(
+              `${progressKeyPrefix}${progress.phase}:`,
+            ),
+          )
+        ) {
+          return 0;
+        }
+        const lastProgressAt = Math.max(
+          currentBase.createdAt.getTime(),
+          ...previousProgress.map((publication) =>
+            publication.createdAt.getTime(),
+          ),
+        );
+        if (
+          currentEvent.createdAt.getTime() - lastProgressAt <
+          SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS
+        ) {
+          return 0;
+        }
+
+        const bindings = await resolveChatOriginPublicationBindings(
+          tx,
+          row.companyId,
+          row.issueId,
+          row.runId,
+        );
+        if (
+          !bindings.some(
+            (binding) =>
+              binding.endpointId === row.endpointId &&
+              binding.conversationId === row.conversationId,
+          )
+        ) {
+          return 0;
+        }
+        if (
+          await hasChatRunOwnedProviderInteraction(tx, {
+            companyId: row.companyId,
+            issueId: row.issueId,
+            runId: row.runId,
+          })
+        ) {
+          return 0;
+        }
+        const explicitlyAuthoredFinal = await tx
+          .select({ id: chatPublications.id })
+          .from(chatPublications)
+          .innerJoin(
+            issueComments,
+            and(
+              eq(issueComments.companyId, chatPublications.companyId),
+              eq(issueComments.id, chatPublications.commentId),
+            ),
+          )
+          .where(
+            and(
+              eq(chatPublications.companyId, row.companyId),
+              eq(chatPublications.endpointId, row.endpointId),
+              eq(chatPublications.conversationId, row.conversationId),
+              eq(chatPublications.issueId, row.issueId),
+              eq(issueComments.authorType, "agent"),
+              eq(issueComments.createdByRunId, row.runId),
+              sql`(
+                ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
+                or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
+              )`,
+            ),
+          )
+          .limit(1);
+        if (explicitlyAuthoredFinal.length > 0) return 0;
+
+        await notifyChatPublicationWork(tx);
+        const insertedRows = await tx
+          .insert(chatPublications)
+          .values({
+            companyId: row.companyId,
+            endpointId: row.endpointId,
+            conversationId: row.conversationId,
+            issueId: row.issueId,
+            idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`,
+            payload: projectSafeChatPublication({
+              classification: "external",
+              source: "safe_milestone",
+              text: progress.text,
+              progressState: "working",
+            }),
+            state: "pending",
+          })
+          .onConflictDoNothing()
+          .returning({ id: chatPublications.id });
+        return insertedRows.length;
+      });
+      inserted += result;
+    }
+    if (rows.length < pageSize) break;
+  }
+  return inserted;
+}

server/src/services/chat-run-publication-policy.ts

@@ -0,0 +1,19 @@
+import { ne, notExists, or, sql } from "drizzle-orm";
+import { chatEndpoints, chatGitHubConfigurations } from "@paperclipai/db";
+
+// Only upgraded GitHub bots have the task-scoped response tools. Legacy
+// connections retain automatic progress and final-answer publication.
+export function allowsAutomaticRunPublication() {
+  return or(ne(chatEndpoints.provider, "github"), notExists(
+    sql`(select 1 from ${chatGitHubConfigurations}
+      where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId}
+        and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`,
+  ));
+}
+
+export function chatRunProjectionWindow(input: { since?: Date; limit?: number }) {
+  return {
+    since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000),
+    limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)),
+  };
+}

server/src/services/chat-run-publications.ts

@@ -1,21 +1,21 @@
+import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js";
+import { enqueueSafeNativeChatProgress } from "./chat-native-progress.js";
+export { enqueueSafeNativeChatProgress } from "./chat-native-progress.js";
 import type { ChatRunMilestoneScope } from "./chat-run-milestone-notifications.js";
 import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import {
   and,
   asc,
-  desc,
   eq,
   gt,
   gte,
   inArray,
   isNull,
-  like,
   ne,
   notExists,
   or,
   sql,
 } from "drizzle-orm";
-import { alias } from "drizzle-orm/pg-core";
 import type { Db } from "@paperclipai/db";
 import {
   agents,
@@ -26,7 +26,6 @@ import {
   chatGitHubReviews,
   chatMessageLinks,
   chatPublications,
-  heartbeatRunEvents,
   heartbeatRuns,
   issueComments,
   issues,
@@ -38,10 +37,6 @@ import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } from "./heartbeat-run-summ
 import { resolveChatOriginPublicationBindings } from "./issues.js";
 import { boundRunIssueIdSql } from "./run-task-binding.js";
 import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js";
-import {
-  SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
-  safeNativeChatProgressForEvent,
-} from "./safe-native-chat-progress.js";
 
 type SafeRunMilestone =
   "queued" | "working" | "waiting_for_input" | "completed" | "failed";
@@ -53,15 +48,6 @@ const OWNERSHIP_ATTENTION_CODES = [
 
 export { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON };
 
-// Only upgraded GitHub bots have the task-scoped response tools. Legacy
-// connections retain automatic progress and final-answer publication.
-function allowsAutomaticRunPublication() {
-  return or(ne(chatEndpoints.provider, "github"), notExists(
-    sql`(select 1 from ${chatGitHubConfigurations}
-      where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId}
-        and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`,
-  ));
-}
 
 /**
  * Heartbeat's presentation resolver may externalize its selected final prose
@@ -210,21 +196,6 @@ type ChatRunMilestoneCandidate = {
   agentName: string;
 };
 
-type SafeNativeChatProgressCandidate = {
-  eventId: number;
-  eventSeq: number;
-  eventType: string;
-  eventCreatedAt: Date;
-  runId: string;
-  agentId: string;
-  issueId: string;
-  companyId: string;
-  endpointId: string;
-  conversationId: string;
-  agentName: string;
-};
-
-const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000;
 
 function milestoneForStatus(
   status: string,
@@ -276,384 +247,6 @@ export function safeMilestoneText(input: {
   }`;
 }
 
-/**
- * Projects a bounded sample of native activity into the existing run working
- * lane. The selector intentionally reads only event identity, type, sequence,
- * and time; native messages and payloads stay inside Paperclip.
- */
-export async function enqueueSafeNativeChatProgress(
-  db: Db,
-  options: { since?: Date; limit?: number } = {},
-): Promise<number> {
-  const input = chatRunProjectionWindow(options);
-  const basePublication = alias(
-    chatPublications,
-    "safe_native_progress_base_publication",
-  );
-  const laterEvent = alias(
-    heartbeatRunEvents,
-    "safe_native_progress_later_event",
-  );
-  const issueIdFromContext = boundRunIssueIdSql();
-  let inserted = 0;
-  let cursor: {
-    eventCreatedAt: Date;
-    runId: string;
-    conversationId: string;
-  } | null = null;
-
-  while (inserted < input.limit) {
-    const pageSize = Math.max(25, Math.min(200, input.limit - inserted));
-    const pageCursor: typeof cursor = cursor;
-    const rows: SafeNativeChatProgressCandidate[] = await db
-      .select({
-        eventId: heartbeatRunEvents.id,
-        eventSeq: heartbeatRunEvents.seq,
-        eventType: heartbeatRunEvents.eventType,
-        eventCreatedAt: heartbeatRunEvents.createdAt,
-        runId: heartbeatRuns.id,
-        agentId: heartbeatRuns.agentId,
-        issueId: chatConversations.issueId,
-        companyId: chatConversations.companyId,
-        endpointId: chatConversations.endpointId,
-        conversationId: chatConversations.id,
-        agentName: agents.name,
-      })
-      .from(heartbeatRunEvents)
-      .innerJoin(
-        heartbeatRuns,
-        and(
-          eq(heartbeatRuns.id, heartbeatRunEvents.runId),
-          eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId),
-          eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId),
-          eq(heartbeatRuns.runtimeMode, "native"),
-          eq(heartbeatRuns.status, "running"),
-        ),
-      )
-      .innerJoin(
-        chatConversations,
-        and(
-          eq(chatConversations.companyId, heartbeatRuns.companyId),
-          sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`,
-          inArray(chatConversations.state, ["active", "waiting"]),
-        ),
-      )
-      .innerJoin(
-        chatEndpoints,
-        and(
-          eq(chatEndpoints.companyId, chatConversations.companyId),
-          eq(chatEndpoints.id, chatConversations.endpointId),
-          eq(chatEndpoints.publicationMode, "automatic"),
-          allowsAutomaticRunPublication(),
-          eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId),
-        ),
-      )
-      .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
-      .innerJoin(
-        basePublication,
-        and(
-          eq(basePublication.companyId, heartbeatRuns.companyId),
-          eq(basePublication.endpointId, chatConversations.endpointId),
-          eq(basePublication.conversationId, chatConversations.id),
-          eq(basePublication.issueId, chatConversations.issueId),
-          sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`,
-          inArray(basePublication.state, [
-            "pending",
-            "retry",
-            "streaming",
-            "published",
-          ]),
-        ),
-      )
-      .where(
-        and(
-          inArray(heartbeatRunEvents.eventType, [
-            ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
-          ]),
-          gte(heartbeatRunEvents.createdAt, input.since),
-          sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`,
-          notExists(
-            db
-              .select({ id: laterEvent.id })
-              .from(laterEvent)
-              .where(
-                and(
-                  eq(laterEvent.companyId, heartbeatRunEvents.companyId),
-                  eq(laterEvent.runId, heartbeatRunEvents.runId),
-                  gt(laterEvent.seq, heartbeatRunEvents.seq),
-                  inArray(laterEvent.eventType, [
-                    ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
-                  ]),
-                ),
-              ),
-          ),
-          pageCursor
-            ? or(
-                gt(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt),
-                and(
-                  eq(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt),
-                  or(
-                    gt(heartbeatRuns.id, pageCursor.runId),
-                    and(
-                      eq(heartbeatRuns.id, pageCursor.runId),
-                      gt(chatConversations.id, pageCursor.conversationId),
-                    ),
-                  ),
-                ),
-              )
-            : undefined,
-        ),
-      )
-      .orderBy(
-        asc(heartbeatRunEvents.createdAt),
-        asc(heartbeatRuns.id),
-        asc(chatConversations.id),
-      )
-      .limit(pageSize);
-    if (rows.length === 0) break;
-    const lastRow = rows.at(-1)!;
-    cursor = {
-      eventCreatedAt: lastRow.eventCreatedAt,
-      runId: lastRow.runId,
-      conversationId: lastRow.conversationId,
-    };
-
-    for (const row of rows) {
-      if (inserted >= input.limit) break;
-      const result = await db.transaction(async (tx) => {
-        // Keep the normal issue -> run order used by native finalization. The
-        // publication insert takes an issue FK lock, so locking the run first
-        // would invert that order against a concurrent terminal commit. A
-        // contended issue/run is optional progress, not a reason to hold other
-        // chats: skip it now and revisit it with a fresh cursor next sweep.
-        const currentIssue = await tx
-          .select({ id: issues.id })
-          .from(issues)
-          .where(
-            and(
-              eq(issues.id, row.issueId),
-              eq(issues.companyId, row.companyId),
-            ),
-          )
-          .for("update", { skipLocked: true })
-          .then((currentRows) => currentRows[0] ?? null);
-        if (!currentIssue) return 0;
-        // Native event admission serializes on the run row. Taking the same
-        // short lock makes terminal status, a later event, or a final comment
-        // win before this non-terminal projection can be inserted.
-        const currentRun = await tx
-          .select({ id: heartbeatRuns.id })
-          .from(heartbeatRuns)
-          .where(
-            and(
-              eq(heartbeatRuns.id, row.runId),
-              eq(heartbeatRuns.companyId, row.companyId),
-              eq(heartbeatRuns.agentId, row.agentId),
-              eq(heartbeatRuns.runtimeMode, "native"),
-              eq(heartbeatRuns.status, "running"),
-            ),
-          )
-          .for("update", { skipLocked: true })
-          .then((currentRows) => currentRows[0] ?? null);
-        if (!currentRun) return 0;
-
-        const destination = await tx
-          .select({ id: chatConversations.id })
-          .from(chatConversations)
-          .innerJoin(
-            chatEndpoints,
-            and(
-              eq(chatEndpoints.companyId, chatConversations.companyId),
-              eq(chatEndpoints.id, chatConversations.endpointId),
-              eq(chatEndpoints.publicationMode, "automatic"),
-              allowsAutomaticRunPublication(),
-              eq(chatEndpoints.assignedAgentId, row.agentId),
-            ),
-          )
-          .where(
-            and(
-              eq(chatConversations.companyId, row.companyId),
-              eq(chatConversations.id, row.conversationId),
-              eq(chatConversations.endpointId, row.endpointId),
-              eq(chatConversations.issueId, row.issueId),
-              inArray(chatConversations.state, ["active", "waiting"]),
-            ),
-          )
-          .limit(1)
-          .then((currentRows) => currentRows[0] ?? null);
-        if (!destination) return 0;
-
-        const [currentEvent] = await tx
-          .select({
-            id: heartbeatRunEvents.id,
-            seq: heartbeatRunEvents.seq,
-            eventType: heartbeatRunEvents.eventType,
-            createdAt: heartbeatRunEvents.createdAt,
-          })
-          .from(heartbeatRunEvents)
-          .where(
-            and(
-              eq(heartbeatRunEvents.companyId, row.companyId),
-              eq(heartbeatRunEvents.runId, row.runId),
-              inArray(heartbeatRunEvents.eventType, [
-                ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
-              ]),
-            ),
-          )
-          .orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id))
-          .limit(1);
-        if (
-          !currentEvent ||
-          currentEvent.id !== row.eventId ||
-          currentEvent.seq !== row.eventSeq ||
-          currentEvent.eventType !== row.eventType
-        ) {
-          return 0;
-        }
-        const progress = safeNativeChatProgressForEvent(
-          currentEvent.eventType,
-          row.agentName,
-        );
-        if (!progress) return 0;
-
-        const baseKey = `run:${row.runId}:working:${row.endpointId}`;
-        const [currentBase] = await tx
-          .select({ createdAt: chatPublications.createdAt })
-          .from(chatPublications)
-          .where(
-            and(
-              eq(chatPublications.companyId, row.companyId),
-              eq(chatPublications.endpointId, row.endpointId),
-              eq(chatPublications.conversationId, row.conversationId),
-              eq(chatPublications.issueId, row.issueId),
-              eq(chatPublications.idempotencyKey, baseKey),
-              inArray(chatPublications.state, [
-                "pending",
-                "retry",
-                "streaming",
-                "published",
-              ]),
-            ),
-          )
-          .limit(1);
-        if (!currentBase) return 0;
-
-        const progressKeyPrefix = `${baseKey}:native:`;
-        const previousProgress = await tx
-          .select({
-            createdAt: chatPublications.createdAt,
-            idempotencyKey: chatPublications.idempotencyKey,
-          })
-          .from(chatPublications)
-          .where(
-            and(
-              eq(chatPublications.companyId, row.companyId),
-              eq(chatPublications.endpointId, row.endpointId),
-              eq(chatPublications.conversationId, row.conversationId),
-              eq(chatPublications.issueId, row.issueId),
-              like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`),
-            ),
-          );
-        if (
-          previousProgress.some((publication) =>
-            publication.idempotencyKey.startsWith(
-              `${progressKeyPrefix}${progress.phase}:`,
-            ),
-          )
-        ) {
-          return 0;
-        }
-        const lastProgressAt = Math.max(
-          currentBase.createdAt.getTime(),
-          ...previousProgress.map((publication) =>
-            publication.createdAt.getTime(),
-          ),
-        );
-        if (
-          currentEvent.createdAt.getTime() - lastProgressAt <
-          SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS
-        ) {
-          return 0;
-        }
-
-        const bindings = await resolveChatOriginPublicationBindings(
-          tx,
-          row.companyId,
-          row.issueId,
-          row.runId,
-        );
-        if (
-          !bindings.some(
-            (binding) =>
-              binding.endpointId === row.endpointId &&
-              binding.conversationId === row.conversationId,
-          )
-        ) {
-          return 0;
-        }
-        if (
-          await hasChatRunOwnedProviderInteraction(tx, {
-            companyId: row.companyId,
-            issueId: row.issueId,
-            runId: row.runId,
-          })
-        ) {
-          return 0;
-        }
-        const explicitlyAuthoredFinal = await tx
-          .select({ id: chatPublications.id })
-          .from(chatPublications)
-          .innerJoin(
-            issueComments,
-            and(
-              eq(issueComments.companyId, chatPublications.companyId),
-              eq(issueComments.id, chatPublications.commentId),
-            ),
-          )
-          .where(
-            and(
-              eq(chatPublications.companyId, row.companyId),
-              eq(chatPublications.endpointId, row.endpointId),
-              eq(chatPublications.conversationId, row.conversationId),
-              eq(chatPublications.issueId, row.issueId),
-              eq(issueComments.authorType, "agent"),
-              eq(issueComments.createdByRunId, row.runId),
-              sql`(
-                ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
-                or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
-              )`,
-            ),
-          )
-          .limit(1);
-        if (explicitlyAuthoredFinal.length > 0) return 0;
-
-        await notifyChatPublicationWork(tx);
-        const insertedRows = await tx
-          .insert(chatPublications)
-          .values({
-            companyId: row.companyId,
-            endpointId: row.endpointId,
-            conversationId: row.conversationId,
-            issueId: row.issueId,
-            idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`,
-            payload: projectSafeChatPublication({
-              classification: "external",
-              source: "safe_milestone",
-              text: progress.text,
-              progressState: "working",
-            }),
-            state: "pending",
-          })
-          .onConflictDoNothing()
-          .returning({ id: chatPublications.id });
-        return insertedRows.length;
-      });
-      inserted += result;
-    }
-    if (rows.length < pageSize) break;
-  }
-  return inserted;
-}
 
 type ChatRunMilestoneCursor = {
   updatedAt: string;
@@ -1092,12 +685,6 @@ type ChatRunProjectionInput = {
   limit?: number;
 };
 
-function chatRunProjectionWindow(input: ChatRunProjectionInput) {
-  return {
-    since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000),
-    limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)),
-  };
-}
 
 /** Explicit combined reconciliation retains lifecycle priority and one shared
  * publication budget. Background owners can reconcile the two duties separately. */