PaperClipAI/server/src/services/native-runtime/native-question-bridge.ts

import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
import { and, eq, inArray, sql } from "drizzle-orm";

import type { Db } from "@paperclipai/db";
import { heartbeatRuns, issueThreadInteractions } from "@paperclipai/db";
import type {
  AskUserQuestionsInteraction,
  RespondIssueThreadInteraction,
} from "@paperclipai/shared";

import { questionSetToAskUserQuestionsPayload } from "@paperclipai/shared";

import type { PrpEvent } from "../../vendor/paperclip-runner/index.js";
import {
  parsePaperclipQuestionSet,
  type PaperclipQuestionSet,
} from "../../vendor/paperclip-runner/index.js";
import { logger } from "../../middleware/logger.js";
import { unprocessable } from "../../errors.js";
import { logActivity } from "../activity-log.js";
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import { questionResponseDeliveryService } from "../question-response-delivery.js";
import type { NativeRunStoreBinding } from "./native-run-coordinator-store.js";

import { parseQuestionInteractionAnswers, parseSavedQuestionInteractionAnswers } from "../question-interaction-answers.js";

const QUESTION_KEY_PREFIX = "paperclip-runner-question:";
const REQUEST_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$/;
export const NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY = "nativeQuestionCancellation";

type QueueCommand = (
  type: string,
  payload?: Record<string, unknown>,
  commandId?: string,
) => { readonly commandId: string; readonly controllerSeq: number } | Promise<{ readonly commandId: string; readonly controllerSeq: number }>;

interface NativeQuestionCommandTarget {
  binding: Pick<NativeRunStoreBinding, "companyId" | "issueId" | "runId" | "agentId">;
  queueCommand: QueueCommand;
}

const activeTargets = new Map<string, NativeQuestionCommandTarget>();

interface NativeQuestionIdentity {
  idempotencyKey?: string | null;
  sourceRunId?: string | null;
  payload: unknown;
}

export interface NativeQuestionAuthorizationIdentity extends NativeQuestionIdentity {
  companyId: string;
  issueId: string;
}

export type NativeQuestionCancellationCause =
  | { kind: "issue_terminal"; issueStatus: string }
  | { kind: "interaction_withdrawn"; interactionId: string }
  | { kind: "interaction_cancelled"; interactionId: string };

type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
type NativeQuestionMutationDb = Pick<Db | DbTransaction, "select" | "update">;

function record(value: unknown): Record<string, unknown> | null {
  return value && typeof value === "object" && !Array.isArray(value)
    ? value as Record<string, unknown>
    : null;
}

function requestIdForInteraction(
  interaction: NativeQuestionIdentity,
): string | null {
  const payload = record(interaction.payload);
  if (!interaction.sourceRunId || !payload?.questionSet) return null;
  const key = interaction.idempotencyKey;
  const expectedPrefix = `${QUESTION_KEY_PREFIX}${interaction.sourceRunId}:`;
  if (!key?.startsWith(expectedPrefix)) return null;
  const requestId = key.slice(expectedPrefix.length);
  if (!REQUEST_ID_PATTERN.test(requestId)) return null;
  return typeof payload.runtimeRequestId === "string" && payload.runtimeRequestId !== requestId
    ? null
    : requestId;
}

function toInteractionPayload(questionSet: PaperclipQuestionSet, runtimeRequestId: string) {
  return {
    ...questionSetToAskUserQuestionsPayload(questionSet),
    runtimeRequestId,
    // A generic task comment cannot satisfy this provider request.
    supersedeOnUserComment: false,
  };
}


async function authorizedNativeRun(
  db: Pick<Db | DbTransaction, "select">,
  interaction: NativeQuestionAuthorizationIdentity,
) {
  const requestId = requestIdForInteraction(interaction);
  if (!requestId || !interaction.sourceRunId) return null;
  const run = await db.select({
    id: heartbeatRuns.id,
    companyId: heartbeatRuns.companyId,
    issueId: heartbeatRuns.nativeIssueId,
    agentId: heartbeatRuns.agentId,
    runtimeMode: heartbeatRuns.runtimeMode,
    status: heartbeatRuns.status,
  }).from(heartbeatRuns).where(and(
    eq(heartbeatRuns.id, interaction.sourceRunId),
    eq(heartbeatRuns.companyId, interaction.companyId),
    eq(heartbeatRuns.nativeIssueId, interaction.issueId),
    eq(heartbeatRuns.runtimeMode, "native"),
  )).limit(1).then((rows) => rows[0] ?? null);
  return run ? { ...run, requestId } : null;
}

/** Materialize questions; permission requests use the privileged runtime card. */
export async function projectNativeRuntimeRequest(input: {
  db: Db;
  binding: Pick<NativeRunStoreBinding, "companyId" | "issueId" | "runId" | "agentId" | "normalizedSessionId" | "runnerSourceInstanceId">;
  event: PrpEvent;
}): Promise<AskUserQuestionsInteraction | null> {
  if (input.event.eventType !== "runtime_request.created") return null;
  if (
    input.event.runId !== input.binding.runId
    || input.event.normalizedSessionId !== input.binding.normalizedSessionId
    || input.event.sourceInstanceId !== input.binding.runnerSourceInstanceId
  ) {
    throw new Error("native_runtime_request_binding_mismatch");
  }
  const request = record(record(input.event.payload)?.request);
  if (
    !request
    || request.schema !== "paperclip.runtime_request.v2"
    || request.status !== "pending"
    || typeof request.requestId !== "string"
    || !REQUEST_ID_PATTERN.test(request.requestId)
  ) {
    throw new Error("native_runtime_request_invalid");
  }
  if (request.requestKind === "permission_approval" && request.type === "permission") {
    const choices = Array.isArray(request.choices) ? request.choices.map(record) : [];
    const supported = new Set(["accept", "accept_for_session", "decline", "cancel"]);
    if (
      typeof request.turnId !== "string"
      || request.turnId !== input.event.turnId
      || (request.itemId != null && typeof request.itemId !== "string")
      || (request.itemId ?? null) !== (input.event.itemId ?? null)
      || typeof request.prompt !== "string"
      || !request.prompt.trim()
      || request.prompt.length > 4000
      || choices.length === 0
      || choices.length > 4
      || choices.some((choice) => !choice
        || typeof choice.key !== "string" || !supported.has(choice.key)
        || typeof choice.label !== "string" || !choice.label.trim() || choice.label.length > 500)
      || new Set(choices.map((choice) => choice?.key)).size !== choices.length
      || (request.details !== undefined && !record(request.details))
    ) {
      throw new Error("native_runtime_permission_invalid");
    }
    // The committed run event itself feeds TaskChatProtocolCard. Its decisions
    // go through the instance-admin runtime-request route and the exact pending
    // turn, not the human-only question-response delivery path. Returning here
    // allows the durable coordinator to acknowledge the event without creating
    // a second, less-privileged interaction or changing any offered choices.
    return null;
  }
  if (request.requestKind !== "runtime" || request.type !== "input") {
    throw new Error("native_runtime_request_invalid");
  }
  const questionSet = parsePaperclipQuestionSet(request.input);
  if (questionSet.questions.some((question) => question.textValidation?.pattern !== undefined)) {
    // JavaScript regular expressions have no execution budget. Provider-authored
    // patterns therefore stay fail-closed until the runner contract supplies a
    // bounded regex dialect rather than exposing the server to catastrophic backtracking.
    throw new Error("native_runtime_question_pattern_unsupported");
  }
  const idempotencyKey = `${QUESTION_KEY_PREFIX}${input.binding.runId}:${request.requestId}`;
  const existing = await input.db.select({ id: issueThreadInteractions.id })
    .from(issueThreadInteractions)
    .where(and(
      eq(issueThreadInteractions.companyId, input.binding.companyId),
      eq(issueThreadInteractions.issueId, input.binding.issueId),
      eq(issueThreadInteractions.idempotencyKey, idempotencyKey),
    ))
    .limit(1)
    .then((rows) => rows[0] ?? null);
  const interaction = await issueThreadInteractionService(input.db).create(
    { id: input.binding.issueId, companyId: input.binding.companyId },
    {
      kind: "ask_user_questions",
      idempotencyKey,
      sourceRunId: input.binding.runId,
      resolverPolicy: "human_only",
      continuationPolicy: "none",
      ...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}),
      ...(typeof request.prompt === "string" ? { summary: request.prompt.slice(0, 1000) } : {}),
      payload: toInteractionPayload(questionSet, request.requestId),
    },
    { agentId: input.binding.agentId, runId: input.binding.runId },
    { supersedePendingSiblingInteractions: false },
  ) as AskUserQuestionsInteraction;
  if (!existing) {
    await logActivity(input.db, {
      companyId: input.binding.companyId,
      actorType: "agent",
      actorId: input.binding.agentId,
      agentId: input.binding.agentId,
      runId: input.binding.runId,
      action: "issue.thread_interaction_created",
      entityType: "issue",
      entityId: input.binding.issueId,
      details: {
        interactionId: interaction.id,
        interactionKind: interaction.kind,
        interactionStatus: interaction.status,
        runtimeMode: "native",
      },
    });
  }
  if (interaction.status === "answered") {
    await deliverNativeQuestionResponseDurably(input.db, interaction);
  }
  return interaction;
}

/** Validate untrusted board input before the existing interaction service persists it. */
export async function validateNativeQuestionResponseInput(
  interaction: AskUserQuestionsInteraction,
  input: RespondIssueThreadInteraction,
): Promise<void> {
  if (!requestIdForInteraction(interaction) || !interaction.payload.questionSet) return;
  try {
    await parseQuestionInteractionAnswers(interaction.payload.questionSet, input.answers, interaction.payload.questions);
  } catch (error) {
    throw unprocessable(
      error instanceof Error ? error.message : "Invalid native question response",
      { code: "invalid_question_response" },
    );
  }
}

/** Queue an answered interaction into the active durable PRP command stream. */
export async function deliverNativeQuestionResponse(
  db: Db,
  interaction: AskUserQuestionsInteraction,
): Promise<"not_native" | "pending" | "queued"> {
  if (interaction.status !== "answered" || !interaction.result || !interaction.payload.questionSet) {
    return "not_native";
  }
  const run = await authorizedNativeRun(db, interaction);
  // A historical question can be answered after its provider turn has ended.
  // Fall through to durable fresh-wake delivery instead of waiting forever for
  // a command target that cannot return for this terminal run.
  if (!run || ["succeeded", "failed", "cancelled", "timed_out"].includes(run.status)) return "not_native";
  const response = parseSavedQuestionInteractionAnswers(interaction.payload.questionSet, interaction.result.answers, interaction.payload.questions);
  const target = activeTargets.get(run.id);
  if (
    !target
    || target.binding.companyId !== run.companyId
    || target.binding.issueId !== run.issueId
    || target.binding.agentId !== run.agentId
  ) {
    return "pending";
  }
  try {
    await target.queueCommand(
      "request.resolve",
      { requestId: run.requestId, response: response as unknown as Record<string, unknown> },
      `question_${interaction.id}`,
    );
    return "queued";
  } catch (error) {
    logger.warn(
      { err: error, runId: run.id, interactionId: interaction.id },
      "native question response remains durable for session recovery",
    );
    return "pending";
  }
}

async function deliverNativeQuestionResponseDurably(
  db: Db,
  interaction: AskUserQuestionsInteraction,
): Promise<void> {
  await questionResponseDeliveryService(db, {
    heartbeat: {
      wakeup: async () => {
        throw new Error("native_question_wake_unreachable");
      },
    } as never,
    resolveNativeQuestion: (candidate) => deliverNativeQuestionResponse(db, candidate),
  }).deliver(interaction.id);
}

export async function flushNativeQuestionResponses(
  db: Db,
  runId: string,
): Promise<void> {
  const target = activeTargets.get(runId);
  if (!target) return;
  const interactions = await issueThreadInteractionService(db).listForIssue(target.binding.issueId);
  for (const interaction of interactions) {
    if (
      interaction.kind === "ask_user_questions"
      && interaction.sourceRunId === runId
      && interaction.status === "answered"
    ) {
      await deliverNativeQuestionResponseDurably(db, interaction);
    }
  }
}

export function registerNativeQuestionCommandTarget(target: NativeQuestionCommandTarget): () => void {
  const existing = activeTargets.get(target.binding.runId);
  if (existing) throw new Error("native_question_command_target_conflict");
  activeTargets.set(target.binding.runId, target);
  return () => {
    if (activeTargets.get(target.binding.runId) === target) {
      activeTargets.delete(target.binding.runId);
    }
  };
}

export async function nativeQuestionRunToCancel(
  db: Db,
  interaction: NativeQuestionAuthorizationIdentity,
): Promise<string | null> {
  const run = await authorizedNativeRun(db, interaction);
  return run && ["queued", "running"].includes(run.status) ? run.id : null;
}

/**
 * Persist cancellation intent in the same transaction that closes the issue.
 * The post-commit fast path and the heartbeat recovery sweep both consume this
 * marker, so process exit or a transient process-termination failure cannot
 * strand a native run after the task closes, even if its question is retained.
 */
export async function requestNativeQuestionRunCancellation(
  db: NativeQuestionMutationDb,
  interaction: NativeQuestionAuthorizationIdentity,
  cause: NativeQuestionCancellationCause,
): Promise<string | null> {
  const run = await authorizedNativeRun(db, interaction);
  if (!run || !["queued", "running"].includes(run.status)) return null;
  const marker = JSON.stringify({
    version: 1,
    issueId: interaction.issueId,
    ...cause,
    requestedAt: new Date().toISOString(),
  });
  await notifyLiveRunRecovery(db, { companyId: interaction.companyId, runId: run.id });
  return db.update(heartbeatRuns).set({
    contextSnapshot: sql`jsonb_set(
      case
        when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
          then ${heartbeatRuns.contextSnapshot}
        else '{}'::jsonb
      end,
      array[${NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY}],
      ${marker}::jsonb,
      true
    )`,
    updatedAt: new Date(),
  }).where(and(
    eq(heartbeatRuns.id, run.id),
    eq(heartbeatRuns.companyId, interaction.companyId),
    eq(heartbeatRuns.nativeIssueId, interaction.issueId),
    eq(heartbeatRuns.runtimeMode, "native"),
    inArray(heartbeatRuns.status, ["queued", "running"]),
  )).returning({ id: heartbeatRuns.id }).then((rows) => rows[0]?.id ?? null);
}

/** Capture the minimum bound identity needed to cancel after the issue transaction commits. */
export function nativeQuestionCancellationIdentity(
  interaction: NativeQuestionAuthorizationIdentity,
): NativeQuestionAuthorizationIdentity | null {
  if (!requestIdForInteraction(interaction)) return null;
  return {
    companyId: interaction.companyId,
    issueId: interaction.issueId,
    sourceRunId: interaction.sourceRunId,
    payload: interaction.payload,
    idempotencyKey: interaction.idempotencyKey,
  };
}

export const nativeQuestionBridgeInternals = {
  resetForTests: () => activeTargets.clear(),
};