From b212886949d22b89144a37e7e0a00d90f7ac64dc Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 11:58:53 -0500 Subject: [PATCH] fix(server): preserve native progress cursor precision and isolate projection Co-Authored-By: Paperclip --- ...-10-25-native-chat-progress-sleep-audit.md | 142 ++++++ .../chat-channels.integration.test.ts | 46 ++ server/src/services/chat-native-progress.ts | 393 ++++++++++++++++ .../services/chat-run-publication-policy.ts | 19 + server/src/services/chat-run-publications.ts | 419 +----------------- 5 files changed, 603 insertions(+), 416 deletions(-) create mode 100644 doc/plans/2026-10-25-native-chat-progress-sleep-audit.md create mode 100644 server/src/services/chat-native-progress.ts create mode 100644 server/src/services/chat-run-publication-policy.ts diff --git a/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md b/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md new file mode 100644 index 00000000000..2750c81a053 --- /dev/null +++ b/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. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 0749c7ef30a..85a1ee7205b 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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) => { diff --git a/server/src/services/chat-native-progress.ts b/server/src/services/chat-native-progress.ts new file mode 100644 index 00000000000..47b87e6a2ee --- /dev/null +++ b/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 { + 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`${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; +} diff --git a/server/src/services/chat-run-publication-policy.ts b/server/src/services/chat-run-publication-policy.ts new file mode 100644 index 00000000000..4fa5eed3879 --- /dev/null +++ b/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)), + }; +} diff --git a/server/src/services/chat-run-publications.ts b/server/src/services/chat-run-publications.ts index 87cbae9242d..8238cbeb41c 100644 --- a/server/src/services/chat-run-publications.ts +++ b/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 { - 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. */