PaperClipAI/server/src/services/native-runtime/native-response-presentation-query.ts

import { QueryBuilder } from "drizzle-orm/pg-core";
import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
import { chatActions, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js";

export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) {
  const db = new QueryBuilder();
  return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
    issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
    endpointId: chatActions.endpointId, principalId: chatActions.principalId,
    // Durable publication retry schedule for an ordinary final task reply.
    publicationNextAttemptAt: sql<string | null>`${heartbeatRuns.resultJson} #>> '{responsePublication,nextAttemptAt}'`.as("publication_next_attempt_at"),
    // Agent Chat conversations keep their completion outbox as the bounded owner.
    taskReply: sql<boolean>`coalesce(${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' in ('done', 'needs_review')
      and (coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = ''), false)`.as("task_reply"),
  }).from(nativeRunFinalizations)
    .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
      eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
    .leftJoin(chatActions, and(eq(chatActions.id, heartbeatRuns.wakeupRequestId),
      eq(chatActions.companyId, heartbeatRuns.companyId)))
    .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId)))
    .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
      eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId),
      eq(nativeRunResults.runId, heartbeatRuns.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId)))
    .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId),
      eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId),
      eq(statusDecisions.runId, heartbeatRuns.id), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId)))
    .where(and(
      eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out"]),
      isNotNull(heartbeatRuns.finishedAt), eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner),
      eq(nativeRunResults.schemaStatus, "accepted"),
      sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`,
      sql`coalesce(${heartbeatRuns.resultJson} #> '{presentationDecision,commentAction}', '"create"'::jsonb) = '"create"'::jsonb`,
      // A reply whose bounded publication retries are exhausted stays visible
      // as a named failure instead of being retried forever.
      sql`${heartbeatRuns.resultJson} #>> '{responsePublication,state}' is distinct from 'failed'`,
      sql`${heartbeatRuns.contextSnapshot}->'skipIssueComment' is distinct from 'true'::jsonb`,
      sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`,
      sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`,
      sql`${nativeRunResults.resultJson} #>> '{terminal,runTerminalState}' = 'succeeded'`,
      sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`,
      or(
        and(eq(heartbeatRuns.status, "succeeded"),
          or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
            and(eq(statusDecisions.reasonCode, "conversation_turn_finished"),
              eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId),
              eq(issues.status, "in_progress"), eq(issues.conversationState, "active"),
              or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)),
              sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issues.id}::text`,
              sql`jsonb_typeof(${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration') = 'number'`,
              sql`${heartbeatRuns.contextSnapshot}->>'conversationSessionGeneration' = ${issues.conversationSessionGeneration}::text`,
              exists(db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
                eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id),
                eq(heartbeatRunEvents.agentId, heartbeatRuns.agentId), eq(heartbeatRunEvents.eventType, "item.completed"),
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,runId}' = ${heartbeatRuns.id}::text`,
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`,
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,turnId}' = ${nativeRunResults.turnId}`,
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,text}' ~ '[^[:space:]]'`,
              ))))),
          or(
            and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`,
              sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`),
            and(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
              sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'needs_review'`,
              sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'needs_review'`),
          )),
        and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`,
          sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`,
          sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,summary}') = 'string'
            and ${nativeRunResults.resultJson} #>> '{result,summary}' ~ '[^[:space:]]'`,
          sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,idempotencyKey}') = 'string'
            and ${nativeRunResults.resultJson} #>> '{result,continuation,idempotencyKey}' ~ '[^[:space:]]'`,
          or(
            and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'response_wake'`,
              sql`${nativeRunResults.resultJson} #> '{result,attentionRequests}' = '[]'::jsonb`,
              sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,summary}') = 'string'
                and ${nativeRunResults.resultJson} #>> '{result,continuation,summary}' ~ '[^[:space:]]'`),
            and(eq(heartbeatRuns.status, "succeeded"), or(
              and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'same_agent'`,
                eq(statusDecisions.reasonCode, "live_continuation_registered")),
              and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'monitor'`,
                inArray(statusDecisions.reasonCode, ["scheduled_monitor_waiting", "scheduled_monitor_dispatched", "scheduled_monitor_preserved"])),
            )),
          )),
      ),
      notExists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and(
        eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId),
        eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId),
        eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId),
        inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
      ))),
      scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined,
      scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
      scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined,
      after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined,
    )).orderBy(asc(heartbeatRuns.id)).limit(100);
}

export function nativeResponsePresentationPendingQuery() {
  return nativeResponsePresentationCandidates();
}