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

import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
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,
  eq,
  gt,
  gte,
  inArray,
  isNull,
  ne,
  notExists,
  or,
  sql,
} from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
  agents,
  chatActions,
  chatConversations,
  chatEndpoints,
  chatGitHubConfigurations,
  chatGitHubReviews,
  chatMessageLinks,
  chatPublications,
  heartbeatRuns,
  issueComments,
  issues,
} from "@paperclipai/db";
import { projectSafeChatPublication } from "./chat-publication-projection.js";
import { safeChatTaskUrl } from "./chat-task-url.js";
import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js";
import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } from "./heartbeat-run-summary.js";
import { resolveChatOriginPublicationBindings } from "./issues.js";
import { boundRunIssueIdSql } from "./run-task-binding.js";
import { chatRunMilestoneSource } from "./chat-run-milestone-candidates.js";
import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js";

type SafeRunMilestone =
  "queued" | "working" | "waiting_for_input" | "completed" | "failed";

const OWNERSHIP_ATTENTION_CODES = [
  "native_execution_ownership_unverified",
  "native_adopted_runner_authentication_timeout",
] as const;

export { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON };


/**
 * Heartbeat's presentation resolver may externalize its selected final prose
 * only when the run has an exact causal path back to a live chat binding.
 * Keeping this decision beside milestone lineage makes the publication and
 * generic-completion paths share the same origin proof.
 */
export async function resolveChatRunPresentationAuthorizationReason(
  db: Db,
  input: { companyId: string; issueId: string; runId: string },
): Promise<
  typeof CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON | "internal_agent_write"
> {
  const [run] = await db
    .select({ resultJson: heartbeatRuns.resultJson })
    .from(heartbeatRuns)
    .where(
      and(
        eq(heartbeatRuns.id, input.runId),
        eq(heartbeatRuns.companyId, input.companyId),
      ),
    )
    .limit(1);
  if (
    run?.resultJson?.finalizationReasonCode === "governed_response_waiting" &&
    !(await authorizeNativeChatReviewPresentation(db, {
      ...input,
      resultJson: run.resultJson,
    }))
  ) {
    return "internal_agent_write";
  }
  const bindings = await resolveChatOriginPublicationBindings(
    db,
    input.companyId,
    input.issueId,
    input.runId,
  );
  if (bindings.length === 0) return "internal_agent_write";
  // A native question/confirmation is the provider-visible result of its
  // originating run. Keep the runner's final presentation as an internal
  // Paperclip comment even if a fast provider answer resolves the interaction
  // before this check; otherwise model metadata can appear as a noisy sibling
  // beside the card or its continuation response.
  if (await hasChatRunOwnedProviderInteraction(db, input)) {
    return "internal_agent_write";
  }
  // GitHub replies are authored through task-scoped tools. Runner-selected
  // final prose stays local even when no tool reply has been sent yet.
  const githubEndpoints = await db
    .select({ id: chatEndpoints.id })
    .from(chatEndpoints)
    .innerJoin(chatGitHubConfigurations, and(
      eq(chatGitHubConfigurations.companyId, chatEndpoints.companyId),
      eq(chatGitHubConfigurations.endpointId, chatEndpoints.id),
    ))
    .where(and(
      eq(chatEndpoints.companyId, input.companyId),
      eq(chatEndpoints.provider, "github"),
      inArray(chatEndpoints.id, bindings.map(binding => binding.endpointId)),
    ));
  if (bindings.every(binding => githubEndpoints.some(endpoint => endpoint.id === binding.endpointId))) {
    return "internal_agent_write";
  }
  return CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON;
}

/** A missing receipt is not proof that a provider write never arrived. */
export async function githubRunReplyState(
  db: Db,
  input: { companyId: string; issueId: string; runId: string; endpointId: string; sourceCommentId?: string; requireSettled?: boolean },
): Promise<"none" | "unsettled" | "confirmed"> {
  const actions = await db
    .select({
      status: chatActions.status,
      operation: sql<string>`${chatActions.payload}->>'operation'`,
      reviewId: sql<string | null>`${chatActions.payload}->>'reviewId'`,
      result: sql<Record<string, unknown> | null>`case when ${chatActions.result} is null then null else jsonb_build_object(
        'reviewId', ${chatActions.result}->'reviewId', 'id', ${chatActions.result}->'id', 'url', ${chatActions.result}->'url', 'summaryUrl', ${chatActions.result}->'summaryUrl',
        'replyWriteStarted', ${chatActions.result}->'replyWriteStarted', 'retryable', ${chatActions.result}->'retryable',
        'code', ${chatActions.result}->'code', 'attempts', ${chatActions.result}->'attempts',
        'deliveryUnknown', ${chatActions.result}->'deliveryUnknown') end`,
    })
    .from(chatActions)
    .innerJoin(chatConversations, and(
      eq(chatConversations.id, chatActions.conversationId),
      eq(chatConversations.companyId, input.companyId),
      eq(chatConversations.endpointId, input.endpointId),
      eq(chatConversations.issueId, input.issueId),
    ))
    .innerJoin(chatEndpoints, and(
      eq(chatEndpoints.id, input.endpointId),
      eq(chatEndpoints.companyId, input.companyId),
      eq(chatEndpoints.provider, "github"),
      sql`${chatEndpoints.assignedAgentId}::text = ${chatActions.payload}->'session'->>'agentId'`,
    ))
    .where(and(
      eq(chatActions.companyId, input.companyId),
      eq(chatActions.endpointId, input.endpointId),
      eq(chatActions.kind, "github_review_publication"),
      sql`${chatActions.payload}->'session'->>'companyId' = ${input.companyId}`,
      sql`${chatActions.payload}->'session'->>'issueId' = ${input.issueId}`,
      sql`${chatActions.payload}->'session'->>'runId' = ${input.runId}`,
      sql`${chatActions.payload}->>'operation' in ('comment', 'formal_review', 'assessment')`,
      input.sourceCommentId ? sql`exists (select 1 from chat_message_links source_link
        where source_link.company_id = ${chatActions.companyId} and source_link.endpoint_id = ${chatActions.endpointId}
          and source_link.conversation_id = ${chatActions.conversationId} and source_link.delivery_id = ${chatActions.deliveryId}
          and source_link.direction = 'inbound' and source_link.comment_id = ${input.sourceCommentId}::uuid)` : undefined,
    ));
  let unsettled = false, confirmed = false;
  for (const action of actions) {
    if (input.requireSettled && (action.status !== "processed" || action.result?.deliveryUnknown === true)) {
      unsettled = true;
    }
    // An assessment can publish findings before a later step fails or loses
    // authority. Those durable receipts still prove a reply was delivered.
    if (action.operation === "assessment" && !input.requireSettled) {
      const [review] = await db
        .select({ receipts: chatGitHubReviews.publicationReceipts })
        .from(chatGitHubReviews)
        .where(and(
          eq(chatGitHubReviews.companyId, input.companyId),
          eq(chatGitHubReviews.endpointId, input.endpointId),
          eq(chatGitHubReviews.runId, input.runId),
          eq(chatGitHubReviews.issueId, input.issueId),
          sql`${chatGitHubReviews.id}::text = ${String(action.reviewId ?? action.result?.reviewId ?? "")}`,
        ));
      if (review && Object.values(review.receipts).some(receipt => receipt.id && receipt.url)) return "confirmed";
    }
    if (action.status === "processed") {
      if (action.operation === "comment" || action.operation === "formal_review") {
        if (action.result?.id && action.result?.url) {
          if (!input.requireSettled) return "confirmed";
          confirmed = true;
        } else unsettled = true;
      }
      if (action.operation === "assessment") {
        if (action.result?.summaryUrl) {
          if (!input.requireSettled) return "confirmed";
          confirmed = true;
        } else if (input.requireSettled) unsettled = true;
      }
    } else if (action.result?.replyWriteStarted === true || ["received", "processing"].includes(action.status) ||
      (action.status === "failed" && (action.result?.retryable === true || action.result?.code === "publication_failed")) ||
      // A later authorization denial cannot disprove an earlier ambiguous write.
      (action.status === "cancelled" && Number(action.result?.attempts ?? 0) > 1)) {
      unsettled = true;
    }
  }
  return unsettled ? "unsettled" : confirmed ? "confirmed" : "none";
}

type ChatRunMilestoneCandidate = {
  runId: string;
  runStatus: string;
  runErrorCode: string | null;
  runUpdatedAt: string;
  issueId: string;
  companyId: string;
  endpointId: string;
  conversationId: string;
  agentName: string;
};


function milestoneForStatus(
  status: string,
  errorCode: string | null,
): SafeRunMilestone | null {
  if (status === "queued") return "queued";
  if (
    status === "running" &&
    OWNERSHIP_ATTENTION_CODES.some((code) => code === errorCode)
  )
    return "waiting_for_input";
  if (status === "running") return "working";
  if (status === "succeeded") return "completed";
  if (["interrupted", "failed", "timed_out", "cancelled"].includes(status))
    return "failed";
  return null;
}

export function safeMilestoneText(input: {
  agentName: string;
  errorCode?: string | null;
  milestone: SafeRunMilestone;
  issueId: string;
  publicBaseUrl?: string | null;
}): string {
  if (input.milestone === "queued") return `${input.agentName} is queued.`;
  if (input.milestone === "working") return `${input.agentName} is working…`;
  if (input.errorCode === "slack_session_stopped")
    return `${input.agentName} stopped at your request.`;
  const taskUrl = safeChatTaskUrl(input.publicBaseUrl, input.issueId);
  const recovery =
    input.milestone === "completed"
      ? `${input.agentName} has no final response to show yet.`
      : input.milestone === "waiting_for_input"
      ? `${input.agentName} needs a Paperclip admin to safely recover this turn before more work can start.`
      : input.errorCode === "low_trust_isolation_unavailable"
        ? `${input.agentName} couldn't safely start this turn because this task was started for an unlinked external guest and isolated guest execution isn't available. Ask a Paperclip admin to create a private identity link for this account or enable isolated guest execution, then start a new task.`
        : input.errorCode === "native_provider_usage_limit"
          ? `${input.agentName} couldn't complete this turn because the model provider's usage allowance is exhausted. A Paperclip admin needs to restore capacity before retrying.`
          : input.errorCode === "native_event_replay_conflict"
            ? `${input.agentName} couldn't safely continue this turn. A Paperclip admin needs to review the run before it can be retried.`
            : input.errorCode === "native_session_cleanup_quarantined"
              ? `${input.agentName} couldn't start this turn because an earlier session needs recovery. Your request is saved. Ask a Paperclip admin to recover that session before retrying; sending the request again won't repair it.`
              : `${input.agentName} stopped before completing this turn.`;
  return `${recovery}${
    taskUrl
      ? ` Open the task in Paperclip: ${taskUrl}`
      : " Open the task in Paperclip for details."
  }`;
}


type ChatRunMilestoneCursor = {
  updatedAt: string;
  runId: string;
  conversationId: string;
};

const explicitlyAuthoredCommentReason = sql<boolean>`(
    ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
    or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
  )`;

function lifecycleMilestonePages(db: Db, since: Date, scope?: ChatRunMilestoneScope) {
  const issueIdFromContext = boundRunIssueIdSql();
  const milestoneFromStatus = sql<string>`case
    when ${heartbeatRuns.status} = 'queued' then 'queued'
    when ${heartbeatRuns.status} = 'running'
      and ${inArray(heartbeatRuns.errorCode, [...OWNERSHIP_ATTENTION_CODES])}
      then 'waiting_for_input'
    when ${heartbeatRuns.status} = 'running' then 'working'
    when ${heartbeatRuns.status} = 'succeeded' then 'completed'
    else 'failed'
  end`;
  const hasQuestionContinuationTarget = sql<boolean>`exists (
    select 1
    from issue_question_response_deliveries question_delivery
    inner join issue_thread_interactions question_interaction
      on question_interaction.company_id = question_delivery.company_id
      and question_interaction.issue_id = question_delivery.issue_id
      and question_interaction.id = question_delivery.interaction_id
    inner join chat_publications question_prompt
      on question_prompt.company_id = question_delivery.company_id
      and question_prompt.issue_id = question_delivery.issue_id
      and question_prompt.payload ->> 'interactionId' = question_delivery.interaction_id::text
      and question_prompt.idempotency_key = 'interaction:' || question_delivery.interaction_id::text || ':' || question_prompt.endpoint_id::text
      and question_prompt.state = 'published'
    where question_delivery.company_id = ${heartbeatRuns.companyId}
      and question_delivery.issue_id::text = ${issueIdFromContext}
      and question_delivery.target_run_id = ${heartbeatRuns.id}
      and question_delivery.status in ('delivered', 'fallback_queued')
      and question_interaction.kind = 'ask_user_questions'
      and question_interaction.status = 'answered'
  )`;
  const hasDirectInteractionContinuation = sql<boolean>`exists (
    select 1
    from issue_thread_interactions interaction
    inner join chat_publications interaction_prompt
      on interaction_prompt.company_id = interaction.company_id
      and interaction_prompt.issue_id = interaction.issue_id
      and interaction_prompt.payload ->> 'interactionId' = interaction.id::text
      and interaction_prompt.idempotency_key = 'interaction:' || interaction.id::text || ':' || interaction_prompt.endpoint_id::text
      and interaction_prompt.state = 'published'
    inner join agent_wakeup_requests continuation_wake
      on continuation_wake.company_id = interaction.company_id
      and continuation_wake.run_id = ${heartbeatRuns.id}
      and continuation_wake.agent_id = ${heartbeatRuns.agentId}
      and continuation_wake.status <> 'skipped'
      and (
        (interaction.kind = 'request_item_verdicts'
          and continuation_wake.payload->>'interactionId' = interaction.id::text
          and continuation_wake.payload->>'sourceRunId' = interaction.source_run_id::text
          and continuation_wake.reason = 'issue_commented')
        or continuation_wake.idempotency_key = case
        when interaction.kind = 'ask_user_questions'
          and interaction.status = 'answered'
          then 'question-response:' || interaction.id::text
        else 'interaction:' || interaction.id::text || ':' || interaction.status
      end)
    where interaction.company_id = ${heartbeatRuns.companyId}
      and interaction.issue_id::text = ${issueIdFromContext}
      and interaction.id::text = ${heartbeatRuns.contextSnapshot} ->> 'interactionId'
      and interaction.source_run_id::text = ${heartbeatRuns.contextSnapshot} ->> 'sourceRunId'
      and (
        (
          interaction.kind = 'ask_user_questions'
          and interaction.status in ('answered', 'cancelled')
        )
        or (
          interaction.kind in ('request_confirmation', 'request_checkbox_confirmation', 'suggest_tasks')
          and interaction.status in ('accepted', 'rejected', 'cancelled')
        )
        or (
          interaction.kind = 'request_item_verdicts'
          and interaction.status in ('pending', 'answered', 'cancelled')
        )
      )
  )`;
  const hasConnectionContinuation = sql<boolean>`(
    ${heartbeatRuns.contextSnapshot}->>'source' = 'connection_intent.resolved' and exists (
      select 1 from connection_intent_deliveries delivery
      inner join issue_thread_interactions interaction on interaction.id = delivery.interaction_id
        and interaction.company_id = delivery.company_id
      inner join agent_wakeup_requests wake on wake.company_id = delivery.company_id
        and wake.run_id = ${heartbeatRuns.id} and wake.agent_id = ${heartbeatRuns.agentId}
        and wake.idempotency_key = 'connection-intent:' || interaction.id::text || ':' || interaction.status
        and wake.status <> 'skipped'
      where delivery.company_id = ${heartbeatRuns.companyId}
        and interaction.issue_id::text = ${issueIdFromContext}
        and interaction.id::text = ${heartbeatRuns.contextSnapshot}->>'interactionId'
    )
  )`;
  const hasMonitorContinuation = sql<boolean>`(
    ${heartbeatRuns.contextSnapshot}->>'source' = 'issue.monitor' and exists (
      select 1 from agent_wakeup_requests monitor_wake
      where monitor_wake.company_id = ${heartbeatRuns.companyId}
        and monitor_wake.run_id = ${heartbeatRuns.id}
        and monitor_wake.agent_id = ${heartbeatRuns.agentId}
        and monitor_wake.reason = 'issue_monitor_due'
        and monitor_wake.status <> 'skipped'
        and monitor_wake.payload->>'issueId' = ${issueIdFromContext}
    )
  )`;
  // This is only a candidate prefilter. The origin resolver below requires the
  // committed decision/effect ledger before selecting any conversation.
  const hasTaskContinuation = sql<boolean>`(
    ${heartbeatRuns.contextSnapshot}->>'source' = 'native_status_decision' and exists (
      select 1 from agent_wakeup_requests continuation_wake
      where continuation_wake.company_id = ${heartbeatRuns.companyId}
        and continuation_wake.run_id = ${heartbeatRuns.id}
        and continuation_wake.agent_id = ${heartbeatRuns.agentId}
        and continuation_wake.reason = 'issue_status_changed'
        and continuation_wake.status in ('queued', 'claimed', 'completed', 'coalesced')
        and continuation_wake.payload->>'issueId' = ${issueIdFromContext}
    )
  )`;
  return (pageSize: number, pageCursor: ChatRunMilestoneCursor | null): Promise<ChatRunMilestoneCandidate[]> => {
    return db
      .select({
        runId: heartbeatRuns.id,
        runStatus: heartbeatRuns.status,
        runErrorCode: heartbeatRuns.errorCode,
        // Keep PostgreSQL precision: a Date cursor can repeatedly select a
        // rejected page whose updated_at has submillisecond digits.
        runUpdatedAt: sql<string>`${heartbeatRuns.updatedAt}::text`,
        issueId: chatConversations.issueId,
        companyId: chatConversations.companyId,
        endpointId: chatConversations.endpointId,
        conversationId: chatConversations.id,
        agentName: agents.name,
      })
      .from(chatRunMilestoneSource(since))
      .innerJoin(
        chatConversations,
        and(
          eq(chatConversations.companyId, heartbeatRuns.companyId),
          sql`${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"),
          eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId),
        ),
      )
      .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
      .leftJoin(
        chatPublications,
        and(
          eq(chatPublications.companyId, heartbeatRuns.companyId),
          eq(chatPublications.endpointId, chatConversations.endpointId),
          sql`${chatPublications.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':' || ${milestoneFromStatus} || ':' || ${chatConversations.endpointId}::text`,
        ),
      )
      .where(
        and(
          inArray(heartbeatRuns.status, [
            "queued",
            "running",
            "succeeded",
            "interrupted",
            "failed",
            "timed_out",
            "cancelled",
          ]),
          or(allowsAutomaticRunPublication(),
            inArray(heartbeatRuns.status, ["succeeded", "interrupted", "failed", "timed_out", "cancelled"])),
          or(
            and(
              sql`${heartbeatRuns.contextSnapshot} ->> 'source' like 'chat:%'`,
              sql`not (${hasQuestionContinuationTarget})`,
            ),
            and(
              inArray(
                sql<string>`${heartbeatRuns.contextSnapshot} ->> 'source'`,
                [
                  "issue.interaction.respond",
                  "issue.interaction.accept",
                  "issue.interaction.reject",
                  "issue.interaction.verdicts",
                  "issue.interaction.cancel",
                  "issue.interaction.withdraw",
                  "external_chat.interaction.resolve",
                ],
              ),
              hasDirectInteractionContinuation,
            ),
            hasQuestionContinuationTarget,
            hasConnectionContinuation,
            hasMonitorContinuation,
            hasTaskContinuation,
          ),
          // Heartbeat marks a run succeeded before the presentation resolver
          // finishes. Waiting for its durable decision prevents a generic
          // completion from racing an explicitly authored final comment.
          sql`(
            ${heartbeatRuns.status} <> 'succeeded'
            or ${heartbeatRuns.resultJson} -> 'presentationDecision' is not null
          )`,
          sql`(${heartbeatRuns.status} <> 'succeeded' or coalesce(
            ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition',
            ${heartbeatRuns.resultJson}->'acceptedResult'->>'reportedWorkDisposition',
            ${heartbeatRuns.resultJson}->'semanticResult'->'result'->>'reportedWorkDisposition', ''
          ) <> 'yielded')`,
          // A successful or interrupted run's selected final remains its
          // authoritative response. Filter it before LIMIT so a page of settled
          // runs cannot starve later milestones.
          or(
            and(
              ne(heartbeatRuns.status, "succeeded"),
              ne(heartbeatRuns.status, "interrupted"),
            ),
            notExists(
              db
                .select({ id: chatPublications.id })
                .from(chatPublications)
                .innerJoin(
                  issueComments,
                  and(
                    eq(issueComments.companyId, chatPublications.companyId),
                    eq(issueComments.id, chatPublications.commentId),
                  ),
                )
                .where(
                  and(
                    eq(chatPublications.companyId, heartbeatRuns.companyId),
                    eq(
                      chatPublications.endpointId,
                      chatConversations.endpointId,
                    ),
                    eq(chatPublications.conversationId, chatConversations.id),
                    eq(issueComments.authorType, "agent"),
                    eq(issueComments.createdByRunId, heartbeatRuns.id),
                    explicitlyAuthoredCommentReason,
                  ),
                ),
            ),
          ),
          or(
            hasConnectionContinuation,
            hasMonitorContinuation,
            hasTaskContinuation,
            sql`exists (
              select 1
              from ${chatMessageLinks}
              where ${chatMessageLinks.companyId} = ${heartbeatRuns.companyId}
                and ${chatMessageLinks.conversationId} = ${chatConversations.id}
                and ${chatMessageLinks.direction} = 'inbound'
                and (
                  ${chatMessageLinks.commentId}::text = (${heartbeatRuns.contextSnapshot} ->> 'wakeCommentId')
                  or ${chatMessageLinks.commentId}::text = (${heartbeatRuns.contextSnapshot} ->> 'commentId')
                  or (${heartbeatRuns.contextSnapshot} -> 'wakeCommentIds') ? ${chatMessageLinks.commentId}::text
                )
            )`,
            sql`exists (
              select 1
              from chat_publications prompt
              where prompt.company_id = ${heartbeatRuns.companyId}
                and prompt.conversation_id = ${chatConversations.id}
                and prompt.issue_id = ${chatConversations.issueId}
                and prompt.state = 'published'
                and prompt.payload ->> 'interactionId' = ${heartbeatRuns.contextSnapshot} ->> 'interactionId'
                and prompt.idempotency_key = 'interaction:' || (prompt.payload ->> 'interactionId') || ':' || prompt.endpoint_id::text
            )`,
            sql`exists (
              select 1
              from issue_question_response_deliveries target_delivery
              inner join chat_publications target_prompt
                on target_prompt.company_id = target_delivery.company_id
                and target_prompt.issue_id = target_delivery.issue_id
                and target_prompt.conversation_id = ${chatConversations.id}
                and target_prompt.payload ->> 'interactionId' = target_delivery.interaction_id::text
                and target_prompt.idempotency_key = 'interaction:' || target_delivery.interaction_id::text || ':' || target_prompt.endpoint_id::text
                and target_prompt.state = 'published'
              where target_delivery.company_id = ${heartbeatRuns.companyId}
                and target_delivery.target_run_id = ${heartbeatRuns.id}
                and target_delivery.status in ('delivered', 'fallback_queued')
            )`,
          ),
          scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
          scope?.issueId ? eq(chatConversations.issueId, scope.issueId) : undefined,
          gte(heartbeatRuns.updatedAt, since),
          isNull(chatPublications.id),
          pageCursor
            ? sql`(${heartbeatRuns.updatedAt}, ${heartbeatRuns.id}, ${chatConversations.id}) >
                (${pageCursor.updatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)`
            : undefined,
        ),
      )
      .orderBy(
        asc(heartbeatRuns.updatedAt),
        asc(heartbeatRuns.id),
        asc(chatConversations.id),
      )
      .limit(pageSize);
  };
}

function milestoneCursor(row: ChatRunMilestoneCandidate): ChatRunMilestoneCursor {
  return { updatedAt: row.runUpdatedAt, runId: row.runId, conversationId: row.conversationId };
}

/** Reconstruct issue-scoped projection work. Candidate rows are only hints;
 * projection still applies current lineage and provider authority. */
export async function loadChatRunLifecycleMilestoneScopes(
  db: Db,
  scope?: ChatRunMilestoneScope,
  signal?: AbortSignal,
): Promise<Array<{ companyId: string; issueId: string }>> {
  const pages = lifecycleMilestonePages(db, chatRunProjectionWindow({}).since, scope);
  const scopes = new Map<string, { companyId: string; issueId: string }>();
  let cursor: ChatRunMilestoneCursor | null = null;
  for (;;) {
    signal?.throwIfAborted();
    const rows = await pages(100, cursor);
    for (const row of rows) scopes.set(`${row.companyId}:${row.issueId}`, { companyId: row.companyId, issueId: row.issueId });
    if (rows.length < 100) return [...scopes.values()];
    cursor = milestoneCursor(rows[rows.length - 1]);
  }
}

/**
 * Project only coarse run lifecycle into bound external conversations. Raw
 * output, errors, tool events, and reasoning stay in Paperclip. Idempotency is
 * keyed by run, milestone, and endpoint so polling and restarts are harmless.
 */
export async function enqueueChatRunLifecycleMilestones(
  db: Db,
  input: ChatRunProjectionInput & { scope?: ChatRunMilestoneScope; signal?: AbortSignal } = {},
): Promise<number> {
  const { since, limit } = chatRunProjectionWindow(input);
  let inserted = 0;
  const pages = lifecycleMilestonePages(db, since, input.scope);
  let cursor: ChatRunMilestoneCursor | null = null;
  const bindingsCache = new Map<
    string,
    Awaited<ReturnType<typeof resolveChatOriginPublicationBindings>>
  >();
  const providerInteractionCache = new Map<string, boolean>();
  while (inserted < limit) {
    const pageSize = Math.max(25, Math.min(200, limit - inserted));
    input.signal?.throwIfAborted();
    const rows = await pages(pageSize, cursor);
    if (rows.length === 0) break;
    cursor = milestoneCursor(rows[rows.length - 1]);

    for (const row of rows) {
      input.signal?.throwIfAborted();
      if (inserted >= limit) break;
      const milestone = milestoneForStatus(row.runStatus, row.runErrorCode);
      if (!milestone) continue;
      const bindingCacheKey = `${row.companyId}:${row.issueId}:${row.runId}`;
      let bindings = bindingsCache.get(bindingCacheKey);
      if (!bindings) {
        bindings = await resolveChatOriginPublicationBindings(
          db,
          row.companyId,
          row.issueId,
          row.runId,
        );
        bindingsCache.set(bindingCacheKey, bindings);
      }
      if (
        !bindings.some(
          (binding) =>
            binding.endpointId === row.endpointId &&
            binding.conversationId === row.conversationId,
        )
      ) {
        continue;
      }
      if (milestone === "completed" || milestone === "failed") {
        let hasProviderInteraction =
          providerInteractionCache.get(bindingCacheKey);
        if (hasProviderInteraction === undefined) {
          hasProviderInteraction = await hasChatRunOwnedProviderInteraction(
            db,
            {
              companyId: row.companyId,
              issueId: row.issueId,
              runId: row.runId,
            },
          );
          providerInteractionCache.set(bindingCacheKey, hasProviderInteraction);
        }
        if (hasProviderInteraction) continue;
      }
      if (milestone === "completed" || row.runStatus === "interrupted") {
        const explicitlyAuthoredPublication = await db
          .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(issueComments.authorType, "agent"),
              eq(issueComments.createdByRunId, row.runId),
              explicitlyAuthoredCommentReason,
            ),
          )
          .limit(1);
        if (explicitlyAuthoredPublication.length > 0) continue;
      }
      const result = await db.transaction(async (tx) => {
        if (milestone === "working") {
          await notifyChatNativeProgress(tx, { companyId: row.companyId, issueId: row.issueId });
        }
        await notifyChatPublicationWork(tx);
        return tx
          .insert(chatPublications)
          .values({
            companyId: row.companyId,
            endpointId: row.endpointId,
            conversationId: row.conversationId,
            issueId: row.issueId,
            idempotencyKey: `run:${row.runId}:${milestone}:${row.endpointId}`,
            payload: projectSafeChatPublication({
              classification: "external",
              source: "safe_milestone",
              text: safeMilestoneText({
                agentName: row.agentName,
                errorCode: row.runErrorCode,
                milestone,
                issueId: row.issueId,
                publicBaseUrl: input.publicBaseUrl,
              }),
              progressState: milestone,
            }),
            state: "pending",
          })
          .onConflictDoNothing()
          .returning({ id: chatPublications.id });
      });
      inserted += result.length;
    }
    if (rows.length < pageSize) break;
  }
  return inserted;
}

type ChatRunProjectionInput = {
  publicBaseUrl?: string | null;
  since?: Date;
  limit?: number;
};


/** Explicit combined reconciliation retains lifecycle priority and one shared
 * publication budget. Background owners can reconcile the two duties separately. */
export async function enqueueChatRunMilestones(
  db: Db,
  input: ChatRunProjectionInput = {},
): Promise<number> {
  const { since, limit } = chatRunProjectionWindow(input);
  const inserted = await enqueueChatRunLifecycleMilestones(db, { ...input, since, limit });
  if (inserted >= limit) return inserted;
  return inserted + await enqueueSafeNativeChatProgress(db, { since, limit: limit - inserted });
}