import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3";
import { and, asc, eq, gte, isNull, sql } from "drizzle-orm";
import { getDb } from "@thinkwork/database-pg";
import {
  artifacts,
  computers,
  computerEvents,
  computerRunbookRuns,
  computerRunbookTasks,
  computerTasks,
  messages,
  threads,
} from "@thinkwork/database-pg/schema";
import {
  resolveConnectionForUser,
  resolveOAuthTokenDetails,
} from "../oauth-token.js";
import { notifyNewMessage, notifyThreadUpdate } from "../../graphql/notify.js";
import {
  ensureDefaultComputerRunbookSkillsMaterialized,
  ensureMigratedComputerWorkspaceSeeded,
} from "./workspace-seed.js";
import { toGraphqlComputerTask } from "./tasks.js";
import type { MemoryRequestContext } from "../memory/index.js";
import {
  completeRunbookRunFromThreadTurn,
  failRunbookRunFromThreadTurn,
} from "../runbooks/runs.js";
import { resolveMessageAttachmentsForDispatch } from "./thread-cutover.js";
import {
  assembleRequesterContext,
  formatRequesterContextForPrompt,
} from "./requester-context.js";
import { recordThreadActivityForIdleLearning } from "../thread-idle-learning/activity.js";
import {
  attachmentDownloadByteLimit,
  detectAttachmentKind,
  extractAttachmentText,
} from "./attachment-extraction.js";
export {
  buildDraftAppletSourceDigest,
  verifyDraftAppletPromotionProof,
} from "../applets/draft-promotion.js";

const db = getDb();
const s3 = new S3Client({});
const GOOGLE_CALENDAR_SCOPE = "https://www.googleapis.com/auth/calendar";
const ARTIFACT_SAVE_MISSING_MESSAGE =
  "I generated a dashboard draft but could not save it as an Artifact. Please retry; no applet was created.";
const RUNBOOK_HEARTBEAT_STALL_THRESHOLD_MINUTES = 5;

const BUILD_INTENT_ACTION_RE =
  /\b(build|create|generate|make|produce|assemble|compose)\b/i;
const ARTIFACT_INTENT_NOUN_RE =
  /\b(applet|app|artifact|dashboard|report|briefing|interactive\s+(surface|view|tool|dashboard)|surface)\b/i;

export class ComputerNotFoundError extends Error {
  constructor(readonly computerId: string) {
    super(`Computer not found: ${computerId}`);
    this.name = "ComputerNotFoundError";
  }
}

export class ComputerTaskNotFoundError extends Error {
  constructor(readonly taskId: string) {
    super(`Computer task not found: ${taskId}`);
    this.name = "ComputerTaskNotFoundError";
  }
}

export class ComputerTaskDelegationError extends Error {
  constructor(
    message: string,
    readonly statusCode = 400,
  ) {
    super(message);
    this.name = "ComputerTaskDelegationError";
  }
}

export async function resolveComputerRuntimeConfig(input: {
  tenantId: string;
  computerId: string;
}) {
  const computer = await loadComputer(input.tenantId, input.computerId);
  return {
    tenantId: computer.tenant_id,
    computerId: computer.id,
    desiredRuntimeStatus: computer.desired_runtime_status,
    runtimeStatus: computer.runtime_status,
    runtimeConfig: computer.runtime_config,
    liveWorkspaceRoot: computer.live_workspace_root,
    efsAccessPointId: computer.efs_access_point_id,
    ecsServiceName: computer.ecs_service_name,
  };
}

export async function recordComputerHeartbeat(input: {
  tenantId: string;
  computerId: string;
  runtimeStatus: string;
  runtimeVersion?: string | null;
  workspaceRoot?: string | null;
}) {
  const [row] = await db
    .update(computers)
    .set({
      runtime_status: input.runtimeStatus,
      live_workspace_root: input.workspaceRoot ?? undefined,
      last_heartbeat_at: new Date(),
      last_active_at: new Date(),
      updated_at: new Date(),
    })
    .where(
      and(
        eq(computers.tenant_id, input.tenantId),
        eq(computers.id, input.computerId),
      ),
    )
    .returning({
      id: computers.id,
      runtime_status: computers.runtime_status,
      live_workspace_root: computers.live_workspace_root,
      last_heartbeat_at: computers.last_heartbeat_at,
      last_active_at: computers.last_active_at,
    });
  if (!row) throw new ComputerNotFoundError(input.computerId);
  let staleRunbookTasksReconciled = 0;
  try {
    staleRunbookTasksReconciled = await reconcileStaleComputerRunbookTasks({
      tenantId: input.tenantId,
      computerId: input.computerId,
    });
  } catch (err) {
    console.error(
      "[computer-runtime] failed to reconcile stale runbook tasks",
      {
        tenantId: input.tenantId,
        computerId: input.computerId,
        message: err instanceof Error ? err.message : String(err),
      },
    );
  }
  try {
    await ensureMigratedComputerWorkspaceSeeded({
      tenantId: input.tenantId,
      computerId: input.computerId,
    });
  } catch (err) {
    console.error("[computer-runtime] failed to seed migrated workspace", {
      tenantId: input.tenantId,
      computerId: input.computerId,
      message: err instanceof Error ? err.message : String(err),
    });
  }
  try {
    await ensureDefaultComputerRunbookSkillsMaterialized({
      tenantId: input.tenantId,
      computerId: input.computerId,
    });
  } catch (err) {
    console.error(
      "[computer-runtime] failed to materialize default runbook skills",
      {
        tenantId: input.tenantId,
        computerId: input.computerId,
        message: err instanceof Error ? err.message : String(err),
      },
    );
  }
  return {
    computerId: row.id,
    runtimeStatus: row.runtime_status,
    liveWorkspaceRoot: row.live_workspace_root,
    lastHeartbeatAt: row.last_heartbeat_at,
    lastActiveAt: row.last_active_at,
    runtimeVersion: input.runtimeVersion ?? null,
    staleRunbookTasksReconciled,
  };
}

async function reconcileStaleComputerRunbookTasks(input: {
  tenantId: string;
  computerId: string;
}) {
  const result = await db.execute(sql`
    WITH candidates AS (
      SELECT
        task.id AS task_id,
        task.title AS task_title,
        run.id AS run_id,
        run.thread_id,
        thread.title AS thread_title,
        COALESCE(task.started_at, task.updated_at, task.created_at) AS stale_since,
        CASE
          WHEN task.details->'supervision'->>'staleAfterMinutes' ~ '^[0-9]+$'
            THEN LEAST(
              GREATEST((task.details->'supervision'->>'staleAfterMinutes')::int, 1),
              120
            )
          ELSE ${RUNBOOK_HEARTBEAT_STALL_THRESHOLD_MINUTES}::int
        END AS stale_after_minutes
      FROM ${computerRunbookTasks} task
      JOIN ${computerRunbookRuns} run ON run.id = task.run_id
      LEFT JOIN ${threads} thread ON thread.id = run.thread_id
      WHERE task.tenant_id = ${input.tenantId}::uuid
        AND run.computer_id = ${input.computerId}::uuid
        AND task.status = 'running'
        AND run.status IN ('queued', 'running')
    )
    SELECT *
    FROM candidates
    WHERE stale_since <= NOW() - make_interval(mins => stale_after_minutes)
    LIMIT 25
  `);

  const rows = (result.rows || []) as Array<{
    task_id: string;
    task_title: string;
    run_id: string;
    thread_id: string | null;
    thread_title: string | null;
    stale_after_minutes: number;
  }>;
  if (rows.length === 0) return 0;

  let processed = 0;
  for (const row of rows) {
    const staleAfterMinutes =
      typeof row.stale_after_minutes === "number"
        ? row.stale_after_minutes
        : Number(row.stale_after_minutes) ||
          RUNBOOK_HEARTBEAT_STALL_THRESHOLD_MINUTES;
    const message = `Runbook task exceeded the ${staleAfterMinutes} minute execution budget.`;
    await db.execute(sql`
      UPDATE ${computerRunbookTasks}
      SET
        status = 'failed',
        error = jsonb_build_object(
          'code', 'runbook_step_timed_out',
          'message', ${message}::text,
          'source', 'computer_heartbeat',
          'staleAfterMinutes', ${staleAfterMinutes}::int
        ),
        completed_at = NOW(),
        updated_at = NOW()
      WHERE id = ${row.task_id}::uuid
        AND status = 'running'
    `);

    await db.execute(sql`
      UPDATE ${computerRunbookTasks}
      SET
        status = 'skipped',
        completed_at = NOW(),
        updated_at = NOW()
      WHERE run_id = ${row.run_id}::uuid
        AND status = 'pending'
    `);

    await db.execute(sql`
      UPDATE ${computerRunbookRuns}
      SET
        status = 'failed',
        error = jsonb_build_object(
          'code', 'runbook_step_timed_out',
          'message', ${message}::text,
          'source', 'computer_heartbeat',
          'taskId', ${row.task_id}::text,
          'staleAfterMinutes', ${staleAfterMinutes}::int
        ),
        completed_at = NOW(),
        updated_at = NOW()
      WHERE id = ${row.run_id}::uuid
        AND status IN ('queued', 'running')
    `);

    await db.execute(sql`
      UPDATE ${computerTasks}
      SET
        status = 'failed',
        error = jsonb_build_object(
          'code', 'runbook_step_timed_out',
          'message', ${message}::text,
          'source', 'computer_heartbeat',
          'runbookRunId', ${row.run_id}::text
        ),
        completed_at = NOW(),
        updated_at = NOW()
      WHERE tenant_id = ${input.tenantId}::uuid
        AND computer_id = ${input.computerId}::uuid
        AND task_type = 'runbook_execute'
        AND status IN ('pending', 'running')
        AND input->>'runbookRunId' = ${row.run_id}
    `);

    if (row.thread_id) {
      const content = `**Stopped:** ${row.task_title}\n\n${message}`;
      const preview =
        content.length > 240 ? `${content.slice(0, 237)}...` : content;
      const [assistantMessage] = await db
        .insert(messages)
        .values({
          tenant_id: input.tenantId,
          thread_id: row.thread_id,
          role: "assistant",
          content,
          parts: [
            {
              type: "text",
              id: `runbook-timeout:${row.run_id}:${row.task_id}`,
              text: content,
            },
          ],
          sender_type: "computer",
          sender_id: input.computerId,
          metadata: {
            runbookMessageKey: `runbook-timeout:${row.run_id}:${row.task_id}`,
            runbookRunId: row.run_id,
            runbookTaskId: row.task_id,
            runbookProgressKind: "failed",
            source: "computer_heartbeat",
          },
        })
        .returning({ id: messages.id });

      const [thread] = await db
        .update(threads)
        .set({
          status: "blocked",
          last_response_preview: preview,
          updated_at: new Date(),
        })
        .where(
          and(
            eq(threads.tenant_id, input.tenantId),
            eq(threads.id, row.thread_id),
          ),
        )
        .returning({ status: threads.status, title: threads.title });

      if (assistantMessage) {
        await notifyNewMessage({
          messageId: assistantMessage.id,
          threadId: row.thread_id,
          tenantId: input.tenantId,
          role: "assistant",
          content,
          senderType: "computer",
          senderId: input.computerId,
        }).catch(() => {});
      }
      await notifyThreadUpdate({
        threadId: row.thread_id,
        tenantId: input.tenantId,
        status: thread?.status ?? "blocked",
        title: thread?.title ?? row.thread_title ?? "Untitled thread",
      }).catch(() => {});
    }

    processed++;
  }

  return processed;
}

export async function claimNextComputerTask(input: {
  tenantId: string;
  computerId: string;
}) {
  await loadComputer(input.tenantId, input.computerId);
  const [candidate] = await db
    .select({
      id: computerTasks.id,
      task_type: computerTasks.task_type,
      input: computerTasks.input,
      idempotency_key: computerTasks.idempotency_key,
      created_by_user_id: computerTasks.created_by_user_id,
      created_at: computerTasks.created_at,
    })
    .from(computerTasks)
    .where(
      and(
        eq(computerTasks.tenant_id, input.tenantId),
        eq(computerTasks.computer_id, input.computerId),
        eq(computerTasks.status, "pending"),
      ),
    )
    .orderBy(asc(computerTasks.created_at))
    .limit(1);
  if (!candidate) return null;

  const [claimed] = await db
    .update(computerTasks)
    .set({
      status: "running",
      claimed_at: new Date(),
      updated_at: new Date(),
    })
    .where(
      and(
        eq(computerTasks.id, candidate.id),
        eq(computerTasks.status, "pending"),
      ),
    )
    .returning({
      id: computerTasks.id,
      tenant_id: computerTasks.tenant_id,
      computer_id: computerTasks.computer_id,
      task_type: computerTasks.task_type,
      input: computerTasks.input,
      idempotency_key: computerTasks.idempotency_key,
      created_by_user_id: computerTasks.created_by_user_id,
      claimed_at: computerTasks.claimed_at,
      created_at: computerTasks.created_at,
    });
  if (!claimed) return null;
  return {
    id: claimed.id,
    tenantId: claimed.tenant_id,
    computerId: claimed.computer_id,
    taskType: claimed.task_type,
    input: claimed.input,
    idempotencyKey: claimed.idempotency_key,
    createdByUserId: claimed.created_by_user_id,
    claimedAt: claimed.claimed_at,
    createdAt: claimed.created_at,
  };
}

export async function appendComputerTaskEvent(input: {
  tenantId: string;
  computerId: string;
  taskId?: string | null;
  eventType: string;
  level?: string;
  payload?: unknown;
}) {
  await loadComputer(input.tenantId, input.computerId);
  if (input.taskId) {
    await loadTask(input.tenantId, input.computerId, input.taskId);
  }
  const [event] = await db
    .insert(computerEvents)
    .values({
      tenant_id: input.tenantId,
      computer_id: input.computerId,
      task_id: input.taskId ?? null,
      event_type: input.eventType,
      level: input.level ?? "info",
      payload: input.payload ?? null,
    })
    .returning({
      id: computerEvents.id,
      event_type: computerEvents.event_type,
      level: computerEvents.level,
      created_at: computerEvents.created_at,
    });
  return {
    id: event.id,
    eventType: event.event_type,
    level: event.level,
    createdAt: event.created_at,
  };
}

export async function checkGoogleWorkspaceConnection(input: {
  tenantId: string;
  computerId: string;
  requesterUserId?: string | null;
}) {
  const computer = await loadComputer(input.tenantId, input.computerId);
  const requesterUserId = credentialRequesterUserId(input.requesterUserId);
  const checkedAt = new Date().toISOString();
  if (!requesterUserId) {
    return {
      providerName: "google_productivity",
      connected: false,
      tokenResolved: false,
      reason: "requester_user_required",
      checkedAt,
    };
  }
  const connection = await resolveConnectionForUser(
    computer.tenant_id,
    requesterUserId,
    "google_productivity",
  );

  if (!connection) {
    return {
      providerName: "google_productivity",
      connected: false,
      tokenResolved: false,
      reason: "no_active_connection",
      checkedAt,
    };
  }

  const tokenDetails = await resolveOAuthTokenDetails(
    connection.connectionId,
    computer.tenant_id,
    connection.providerId,
  );
  const grantedScopes = tokenDetails?.grantedScopes ?? [];
  const missingScopes = missingGoogleCalendarScopes(grantedScopes);

  return {
    providerName: "google_productivity",
    connected: true,
    tokenResolved: Boolean(tokenDetails?.accessToken),
    connectionId: connection.connectionId,
    grantedScopes,
    missingScopes,
    calendarScopeGranted: missingScopes.length === 0,
    reason: tokenDetails?.accessToken ? null : "token_unavailable_or_expired",
    checkedAt,
  };
}

export async function resolveGoogleWorkspaceCliToken(input: {
  tenantId: string;
  computerId: string;
  requesterUserId?: string | null;
}) {
  const computer = await loadComputer(input.tenantId, input.computerId);
  const requesterUserId = credentialRequesterUserId(input.requesterUserId);
  const checkedAt = new Date().toISOString();
  const base = {
    providerName: "google_productivity",
    checkedAt,
  };
  if (!requesterUserId) {
    return {
      ...base,
      connected: false,
      tokenResolved: false,
      reason: "requester_user_required",
    };
  }
  const connection = await resolveConnectionForUser(
    computer.tenant_id,
    requesterUserId,
    "google_productivity",
  );

  if (!connection) {
    return {
      ...base,
      connected: false,
      tokenResolved: false,
      reason: "no_active_connection",
    };
  }

  const tokenDetails = await resolveOAuthTokenDetails(
    connection.connectionId,
    computer.tenant_id,
    connection.providerId,
  );

  if (!tokenDetails?.accessToken) {
    return {
      ...base,
      connected: true,
      tokenResolved: false,
      connectionId: connection.connectionId,
      reason: "token_unavailable_or_expired",
    };
  }

  return {
    ...base,
    connected: true,
    tokenResolved: true,
    connectionId: connection.connectionId,
    grantedScopes: tokenDetails.grantedScopes,
    missingScopes: missingGoogleCalendarScopes(tokenDetails.grantedScopes),
    reason: null,
    accessToken: tokenDetails.accessToken,
  };
}

export async function executeThreadTurnTask(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
}) {
  const task = await loadTask(input.tenantId, input.computerId, input.taskId);
  if (task.task_type !== "thread_turn") {
    throw new ComputerTaskDelegationError(
      "Only thread_turn tasks can be executed as Thread turns",
    );
  }
  throw new ComputerTaskDelegationError(
    "Legacy Managed Agent thread_turn execution is disabled; use Computer-native context/response endpoints",
    410,
  );
}

export async function loadThreadTurnContext(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
}) {
  const task = await loadTask(input.tenantId, input.computerId, input.taskId);
  if (task.task_type !== "thread_turn") {
    throw new ComputerTaskDelegationError(
      "Only thread_turn tasks can load Thread turn context",
    );
  }

  const payload = threadTurnPayload(task.input);
  const computer = await loadComputer(input.tenantId, input.computerId);
  const { thread, message } = await loadComputerThreadTurn({
    tenantId: input.tenantId,
    computerId: input.computerId,
    threadId: payload.threadId,
    messageId: payload.messageId,
  });
  const attachments = await loadComputerThreadTurnAttachments({
    tenantId: input.tenantId,
    threadId: payload.threadId,
    metadata: message.metadata,
  });

  const history = await db
    .select({
      id: messages.id,
      role: messages.role,
      content: messages.content,
    })
    .from(messages)
    .where(
      and(
        eq(messages.tenant_id, input.tenantId),
        eq(messages.thread_id, payload.threadId),
      ),
    )
    .orderBy(asc(messages.created_at));

  const requesterUserId =
    payload.requesterUserId ?? task.created_by_user_id ?? null;
  const contextClass =
    payload.contextClass ?? (requesterUserId ? "user" : "system");
  const requesterContext = await assembleRequesterContext({
    tenantId: input.tenantId,
    computerId: input.computerId,
    requesterUserId,
    prompt: message.content ?? "",
    sourceSurface: payload.source,
    contextClass,
    credentialSubject: payload.credentialSubject,
    event: payload.event,
  });
  const requesterContextOverlay =
    formatRequesterContextForPrompt(requesterContext);

  return {
    taskId: input.taskId,
    source: payload.source,
    computerScope: "shared",
    requester: {
      userId: requesterUserId,
      actorType: payload.actorType,
      actorId: payload.actorId,
      contextClass,
    },
    surfaceContext: payload.surfaceContext,
    requesterContext,
    requesterContextOverlay,
    computer: {
      id: computer.id,
      name: computer.name,
      slug: computer.slug,
      workspaceRoot: computer.live_workspace_root ?? "/workspace",
    },
    thread: {
      id: thread.id,
      title: thread.title,
    },
    message: {
      id: message.id,
      content: message.content ?? "",
    },
    attachments,
    messagesHistory: history
      .filter((row) => row.role === "user" || row.role === "assistant")
      .map((row) => ({
        id: row.id,
        role: row.role,
        content: row.content ?? "",
      })),
    model: resolveComputerChatModel(computer.runtime_config),
    systemPrompt: buildComputerThreadSystemPrompt({
      computer,
      threadTitle: thread.title,
      requesterUserId,
      requesterContext,
    }),
  };
}

export async function recordThreadTurnResponse(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
  content: string;
  model?: string | null;
  source?: string | null;
  usage?: unknown;
}) {
  const task = await loadTask(input.tenantId, input.computerId, input.taskId);
  if (task.task_type !== "thread_turn") {
    throw new ComputerTaskDelegationError(
      "Only thread_turn tasks can record Thread responses",
    );
  }

  const payload = threadTurnPayload(task.input);
  if (input.source && input.source !== payload.source) {
    throw new ComputerTaskDelegationError(
      `Thread response source ${input.source} does not match task source ${payload.source}`,
    );
  }
  const { thread, message } = await loadComputerThreadTurn({
    tenantId: input.tenantId,
    computerId: input.computerId,
    threadId: payload.threadId,
    messageId: payload.messageId,
  });
  const requestedContent = input.content;
  const buildStylePrompt = isBuildStyleArtifactPrompt(message.content ?? "");
  const directSaveAppSucceeded = hasSuccessfulDirectSaveAppEvidence(
    input.usage,
  );
  const draftPreviewParts = draftPreviewPartsFromUsage(input.usage);
  const draftPreviewSucceeded = hasSuccessfulDraftPreviewEvidence(input.usage);
  const initialContent =
    buildStylePrompt && !directSaveAppSucceeded && !draftPreviewSucceeded
      ? ARTIFACT_SAVE_MISSING_MESSAGE
      : requestedContent;
  const assistantMetadata = {
    computerId: input.computerId,
    taskId: input.taskId,
    sourceMessageId: message.id,
    source: payload.source,
    requesterUserId: payload.requesterUserId ?? task.created_by_user_id ?? null,
    contextClass:
      payload.requesterUserId || task.created_by_user_id ? "user" : "system",
    surfaceContext: payload.surfaceContext,
    model: input.model ?? null,
    usage: input.usage ?? null,
    draftPreviewCount: draftPreviewParts.length,
    draftPreviewSucceeded,
  };
  const [assistantMessage] = await db
    .insert(messages)
    .values({
      tenant_id: input.tenantId,
      thread_id: thread.id,
      role: "assistant",
      content: initialContent,
      sender_type: "computer",
      sender_id: input.computerId,
      parts: draftPreviewParts.length > 0 ? draftPreviewParts : undefined,
      metadata: assistantMetadata,
    })
    .returning({ id: messages.id });
  if (!assistantMessage) {
    throw new ComputerTaskDelegationError(
      "Computer thread turn response insert failed",
      500,
    );
  }

  const turnStartedAt =
    task.claimed_at instanceof Date
      ? task.claimed_at
      : task.created_at instanceof Date
        ? task.created_at
        : new Date(0);
  const linkedArtifacts = await db
    .update(artifacts)
    .set({
      source_message_id: assistantMessage.id,
      updated_at: new Date(),
    })
    .where(
      and(
        eq(artifacts.tenant_id, input.tenantId),
        eq(artifacts.thread_id, thread.id),
        eq(artifacts.type, "applet"),
        isNull(artifacts.source_message_id),
        gte(artifacts.created_at, turnStartedAt),
      ),
    )
    .returning({ id: artifacts.id });
  const linkedArtifactIds = linkedArtifacts.map((row) => row.id);
  const artifactSaveMissing =
    buildStylePrompt &&
    !directSaveAppSucceeded &&
    linkedArtifactIds.length === 0 &&
    !draftPreviewSucceeded
      ? {
          reason: "missing_direct_save_app" as const,
          buildStylePrompt: true,
          directSaveAppSucceeded: false,
          draftPreviewSucceeded: false,
          linkedArtifactIds,
        }
      : null;
  const content = artifactSaveMissing
    ? ARTIFACT_SAVE_MISSING_MESSAGE
    : requestedContent;
  if (artifactSaveMissing || content !== initialContent) {
    await db
      .update(messages)
      .set({
        content,
        metadata: {
          ...assistantMetadata,
          artifactSaveMissing,
        },
      })
      .where(
        and(
          eq(messages.tenant_id, input.tenantId),
          eq(messages.id, assistantMessage.id),
        ),
      );
  }

  const preview =
    content.length > 240 ? `${content.slice(0, 237)}...` : content;

  await db
    .update(threads)
    .set({
      last_turn_completed_at: new Date(),
      last_response_preview: preview,
      updated_at: new Date(),
    })
    .where(
      and(eq(threads.tenant_id, input.tenantId), eq(threads.id, thread.id)),
    );

  await appendComputerTaskEvent({
    tenantId: input.tenantId,
    computerId: input.computerId,
    taskId: input.taskId,
    eventType: "thread_turn_response_recorded",
    level: "info",
    payload: {
      threadId: thread.id,
      sourceMessageId: message.id,
      responseMessageId: assistantMessage.id,
      source: payload.source,
      requesterUserId:
        payload.requesterUserId ?? task.created_by_user_id ?? null,
      contextClass:
        payload.requesterUserId || task.created_by_user_id ? "user" : "system",
      surfaceContext: payload.surfaceContext,
      model: input.model ?? null,
      linkedArtifactIds,
      linkedArtifactCount: linkedArtifactIds.length,
      draftPreviewCount: draftPreviewParts.length,
      draftPreviewSucceeded,
      artifactSaveMissing,
    },
  });

  const requesterUserId =
    payload.requesterUserId ?? task.created_by_user_id ?? null;
  if (requesterUserId) {
    const idleLearning = await recordThreadActivityForIdleLearning({
      tenantId: input.tenantId,
      computerId: input.computerId,
      threadId: thread.id,
      requesterUserId,
      source: "assistant_response",
    });
    if (!idleLearning.ok) {
      await appendComputerTaskEvent({
        tenantId: input.tenantId,
        computerId: input.computerId,
        taskId: input.taskId,
        eventType: "idle_memory_learning_schedule_failed",
        level: "warn",
        payload: {
          threadId: thread.id,
          requesterUserId,
          error: idleLearning.error,
        },
      });
    }
  }

  if (payload.runbookRunId) {
    const runbookOutput = {
      responseMessageId: assistantMessage.id,
      threadId: thread.id,
      sourceMessageId: message.id,
      linkedArtifactIds,
      linkedArtifactCount: linkedArtifactIds.length,
      draftPreviewCount: draftPreviewParts.length,
      draftPreviewSucceeded,
      artifactSaveMissing,
    };
    if (artifactSaveMissing) {
      await failRunbookRunFromThreadTurn({
        tenantId: input.tenantId,
        runId: payload.runbookRunId,
        error: {
          message: ARTIFACT_SAVE_MISSING_MESSAGE,
          ...runbookOutput,
        },
      });
    } else {
      await completeRunbookRunFromThreadTurn({
        tenantId: input.tenantId,
        runId: payload.runbookRunId,
        output: runbookOutput,
      });
    }
  }

  await db
    .update(computerTasks)
    .set({
      status: "completed",
      output: {
        responseMessageId: assistantMessage.id,
        threadId: thread.id,
        sourceMessageId: message.id,
        response: content,
        model: input.model ?? null,
        usage: input.usage ?? null,
        linkedArtifactIds,
        linkedArtifactCount: linkedArtifactIds.length,
        draftPreviewCount: draftPreviewParts.length,
        draftPreviewSucceeded,
        artifactSaveMissing,
      },
      completed_at: new Date(),
      updated_at: new Date(),
    })
    .where(
      and(
        eq(computerTasks.tenant_id, input.tenantId),
        eq(computerTasks.computer_id, input.computerId),
        eq(computerTasks.id, input.taskId),
      ),
    );

  await appendComputerTaskEvent({
    tenantId: input.tenantId,
    computerId: input.computerId,
    taskId: input.taskId,
    eventType: "task_completed",
    level: "info",
    payload: {
      responseMessageId: assistantMessage.id,
      threadId: thread.id,
      source: payload.source,
      requesterUserId:
        payload.requesterUserId ?? task.created_by_user_id ?? null,
    },
  });

  await notifyNewMessage({
    messageId: assistantMessage.id,
    threadId: thread.id,
    tenantId: input.tenantId,
    role: "assistant",
    content,
    senderType: "computer",
    senderId: input.computerId,
  });
  await notifyThreadUpdate({
    threadId: thread.id,
    tenantId: input.tenantId,
    status: thread.status,
    title: thread.title,
  });

  return {
    responded: true,
    mode: "computer_native" as const,
    responseMessageId: assistantMessage.id,
    threadId: thread.id,
    messageId: message.id,
    source: payload.source,
    status: "completed",
    model: input.model ?? null,
    linkedArtifactIds,
    linkedArtifactCount: linkedArtifactIds.length,
    draftPreviewCount: draftPreviewParts.length,
    draftPreviewSucceeded,
    artifactSaveMissing,
  };
}

export function isBuildStyleArtifactPrompt(prompt: string) {
  return (
    BUILD_INTENT_ACTION_RE.test(prompt) && ARTIFACT_INTENT_NOUN_RE.test(prompt)
  );
}

function hasSuccessfulDirectSaveAppEvidence(usage: unknown) {
  const invocations = toolInvocationsFromUsage(usage);
  return invocations.some((invocation) => {
    if (invocation.tool_name !== "save_app") return false;
    if (invocation.type === "sub_agent") return false;
    if (!toolInvocationSucceeded(invocation.status)) return false;

    const output = saveAppOutputPayload(invocation);
    return (
      output?.ok === true &&
      output.persisted === true &&
      typeof output.appId === "string" &&
      output.appId.trim().length > 0
    );
  });
}

function hasSuccessfulDraftPreviewEvidence(usage: unknown) {
  const invocations = toolInvocationsFromUsage(usage);
  return invocations.some((invocation) => {
    if (invocation.tool_name !== "preview_app") return false;
    if (invocation.type === "sub_agent") return false;
    if (!toolInvocationSucceeded(invocation.status)) return false;

    return isSuccessfulDraftPreviewOutput(
      draftPreviewOutputPayload(invocation),
    );
  });
}

export function draftPreviewPartsFromUsage(usage: unknown) {
  const invocations = toolInvocationsFromUsage(usage);
  return invocations
    .filter((invocation) => {
      if (invocation.tool_name !== "preview_app") return false;
      if (invocation.type === "sub_agent") return false;
      if (!toolInvocationSucceeded(invocation.status)) return false;
      return Boolean(draftPreviewOutputPayload(invocation));
    })
    .map((invocation, index) => {
      const output = draftPreviewOutputPayload(invocation);
      const toolCallId =
        stringValue(invocation.tool_use_id) ??
        stringValue(invocation.toolCallId) ??
        stringValue(invocation.id) ??
        `preview_app:${index + 1}`;
      return {
        type: "tool-preview_app",
        toolCallId,
        toolName: "preview_app",
        state: "output-available",
        input: isRecord(invocation.input_json)
          ? invocation.input_json
          : undefined,
        output,
      };
    });
}

function draftPreviewOutputPayload(invocation: Record<string, unknown>) {
  const output = isRecord(invocation.output_json)
    ? invocation.output_json
    : parseRecordJson(stringValue(invocation.output_preview));
  if (!output) return null;
  if (output.type !== "draft_app_preview") return null;
  if (!isRecord(output.draft)) return null;
  return output;
}

function isSuccessfulDraftPreviewOutput(
  output: Record<string, unknown> | null,
) {
  if (!output || output.ok !== true) return false;
  const draft = isRecord(output.draft) ? output.draft : null;
  if (!draft) return false;

  const validation = isRecord(draft.validation) ? draft.validation : null;
  if (!validation || validation.ok !== true) return false;
  if (
    typeof draft.sourceDigest !== "string" ||
    !draft.sourceDigest.startsWith("sha256:")
  ) {
    return false;
  }
  if (
    typeof draft.promotionProof !== "string" ||
    !draft.promotionProof.trim()
  ) {
    return false;
  }

  const dataProvenance = isRecord(draft.dataProvenance)
    ? draft.dataProvenance
    : null;
  const dataStatus = stringValue(dataProvenance?.status)?.toLowerCase();
  if (!dataStatus || !["real", "partial", "unavailable"].includes(dataStatus)) {
    return false;
  }

  const shadcnProvenance = isRecord(draft.shadcnProvenance)
    ? draft.shadcnProvenance
    : null;
  const registryDigest = stringValue(shadcnProvenance?.uiRegistryDigest);
  if (!registryDigest?.startsWith("sha256:")) return false;
  const mcpToolCalls = shadcnProvenance?.mcpToolCalls;
  if (!Array.isArray(mcpToolCalls) || mcpToolCalls.length === 0) return false;

  return true;
}

function toolInvocationsFromUsage(usage: unknown) {
  if (!isRecord(usage)) return [];
  const invocations = usage.tool_invocations;
  return Array.isArray(invocations) ? invocations.filter(isRecord) : [];
}

function toolInvocationSucceeded(status: unknown) {
  if (status === undefined || status === null) return true;
  if (typeof status !== "string") return false;
  return ["success", "completed", "ok"].includes(status.toLowerCase());
}

function saveAppOutputPayload(invocation: Record<string, unknown>) {
  if (isRecord(invocation.output_json)) return invocation.output_json;
  return parseRecordJson(stringValue(invocation.output_preview));
}

function parseRecordJson(value: string | null) {
  if (!value) return null;
  try {
    const parsed = JSON.parse(value);
    return isRecord(parsed) ? parsed : null;
  } catch {
    return null;
  }
}

function stringValue(value: unknown) {
  return typeof value === "string" && value.trim().length > 0 ? value : null;
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value && typeof value === "object" && !Array.isArray(value));
}

export async function completeComputerTask(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
  output?: unknown;
}) {
  return finishTask({
    ...input,
    status: "completed",
    output: input.output ?? null,
    error: null,
  });
}

export async function failComputerTask(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
  error: unknown;
}) {
  return finishTask({
    ...input,
    status: "failed",
    output: null,
    error: input.error,
  });
}

export async function cancelComputerTask(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
  output?: unknown;
}) {
  return finishTask({
    ...input,
    status: "cancelled",
    output: input.output ?? null,
    error: null,
  });
}

async function finishTask(input: {
  tenantId: string;
  computerId: string;
  taskId: string;
  status: "completed" | "failed" | "cancelled";
  output: unknown;
  error: unknown;
}) {
  const [row] = await db
    .update(computerTasks)
    .set({
      status: input.status,
      output: input.output,
      error: input.error,
      completed_at: new Date(),
      updated_at: new Date(),
    })
    .where(
      and(
        eq(computerTasks.tenant_id, input.tenantId),
        eq(computerTasks.computer_id, input.computerId),
        eq(computerTasks.id, input.taskId),
      ),
    )
    .returning({
      id: computerTasks.id,
      tenant_id: computerTasks.tenant_id,
      computer_id: computerTasks.computer_id,
      task_type: computerTasks.task_type,
      status: computerTasks.status,
      input: computerTasks.input,
      output: computerTasks.output,
      error: computerTasks.error,
      idempotency_key: computerTasks.idempotency_key,
      claimed_at: computerTasks.claimed_at,
      completed_at: computerTasks.completed_at,
      created_by_user_id: computerTasks.created_by_user_id,
      created_at: computerTasks.created_at,
      updated_at: computerTasks.updated_at,
    });
  if (!row) throw new ComputerTaskNotFoundError(input.taskId);
  await appendComputerTaskEvent({
    tenantId: input.tenantId,
    computerId: input.computerId,
    taskId: input.taskId,
    eventType: `task_${input.status}`,
    level: input.status === "failed" ? "error" : "info",
    payload:
      input.status === "failed"
        ? { error: input.error }
        : input.status === "cancelled"
          ? { output: input.output }
          : undefined,
  });
  return toGraphqlComputerTask(row);
}

async function loadComputer(tenantId: string, computerId: string) {
  const [computer] = await db
    .select()
    .from(computers)
    .where(and(eq(computers.tenant_id, tenantId), eq(computers.id, computerId)))
    .limit(1);
  if (!computer) throw new ComputerNotFoundError(computerId);
  return computer;
}

function credentialRequesterUserId(value: string | null | undefined) {
  return typeof value === "string" && value.trim() ? value.trim() : null;
}

function missingGoogleCalendarScopes(grantedScopes: string[]) {
  return grantedScopes.includes(GOOGLE_CALENDAR_SCOPE)
    ? []
    : [GOOGLE_CALENDAR_SCOPE];
}

async function loadTask(tenantId: string, computerId: string, taskId: string) {
  const [task] = await db
    .select({
      id: computerTasks.id,
      task_type: computerTasks.task_type,
      input: computerTasks.input,
      created_by_user_id: computerTasks.created_by_user_id,
      claimed_at: computerTasks.claimed_at,
      created_at: computerTasks.created_at,
    })
    .from(computerTasks)
    .where(
      and(
        eq(computerTasks.tenant_id, tenantId),
        eq(computerTasks.computer_id, computerId),
        eq(computerTasks.id, taskId),
      ),
    )
    .limit(1);
  if (!task) throw new ComputerTaskNotFoundError(taskId);
  return task;
}

async function loadComputerThreadTurn(input: {
  tenantId: string;
  computerId: string;
  threadId: string;
  messageId: string;
}) {
  const [thread] = await db
    .select({
      id: threads.id,
      title: threads.title,
      status: threads.status,
    })
    .from(threads)
    .where(
      and(
        eq(threads.tenant_id, input.tenantId),
        eq(threads.id, input.threadId),
        eq(threads.computer_id, input.computerId),
      ),
    )
    .limit(1);
  if (!thread) {
    throw new ComputerTaskDelegationError(
      "Computer-owned Thread not found for task",
      409,
    );
  }

  const [message] = await db
    .select({
      id: messages.id,
      content: messages.content,
      role: messages.role,
      metadata: messages.metadata,
    })
    .from(messages)
    .where(
      and(
        eq(messages.tenant_id, input.tenantId),
        eq(messages.thread_id, input.threadId),
        eq(messages.id, input.messageId),
      ),
    )
    .limit(1);
  if (!message) {
    throw new ComputerTaskDelegationError(
      "Thread turn message not found for task",
      409,
    );
  }
  if (message.role !== "user") {
    throw new ComputerTaskDelegationError(
      "Only user messages can trigger Computer thread turns",
      400,
    );
  }

  return { thread, message };
}

async function loadComputerThreadTurnAttachments(input: {
  tenantId: string;
  threadId: string;
  metadata: unknown;
}) {
  const attachments = await resolveMessageAttachmentsForDispatch({
    tenantId: input.tenantId,
    metadata: input.metadata,
  });
  if (attachments.length === 0) return [];

  const expectedPrefix = `tenants/${input.tenantId}/attachments/${input.threadId}/`;
  const bucket = process.env.WORKSPACE_BUCKET || process.env.BUCKET_NAME || "";

  return Promise.all(
    attachments.map(async (attachment) => {
      const base = {
        attachmentId: attachment.attachmentId,
        name: attachment.name,
        mimeType: attachment.mimeType,
        sizeBytes: attachment.sizeBytes,
        readable: false,
        truncated: false,
      };
      if (!attachment.s3Key.startsWith(expectedPrefix)) {
        return { ...base, reason: "prefix_mismatch" };
      }
      const extractionKind = detectAttachmentKind(attachment);
      if (!extractionKind) {
        return { ...base, reason: "unsupported_mime_type" };
      }
      const byteLimit = attachmentDownloadByteLimit(attachment);
      if (extractionKind !== "text" && attachment.sizeBytes > byteLimit) {
        return { ...base, reason: "attachment_too_large", extractionKind };
      }
      if (!bucket) {
        return { ...base, reason: "workspace_bucket_unconfigured" };
      }

      try {
        const object = await s3.send(
          new GetObjectCommand({
            Bucket: bucket,
            Key: attachment.s3Key,
            Range: `bytes=0-${byteLimit - 1}`,
          }),
        );
        const body = await s3BodyToBuffer(object.Body);
        const extracted = await extractAttachmentText({
          name: attachment.name,
          mimeType: attachment.mimeType,
          sizeBytes: attachment.sizeBytes,
          body,
        });
        return { ...base, ...extracted };
      } catch (err) {
        console.error("[computer-runtime] failed to load message attachment", {
          tenantId: input.tenantId,
          threadId: input.threadId,
          attachmentId: attachment.attachmentId,
          message: err instanceof Error ? err.message : String(err),
        });
        return { ...base, reason: "download_failed" };
      }
    }),
  );
}

async function s3BodyToBuffer(body: unknown): Promise<Buffer> {
  if (!body) return Buffer.alloc(0);
  if (Buffer.isBuffer(body)) return body;
  if (body instanceof Uint8Array) return Buffer.from(body);
  if (
    typeof (body as { transformToByteArray?: unknown }).transformToByteArray ===
    "function"
  ) {
    const bytes = await (
      body as { transformToByteArray: () => Promise<Uint8Array> }
    ).transformToByteArray();
    return Buffer.from(bytes);
  }
  throw new Error("Unsupported S3 body type");
}

function resolveComputerChatModel(runtimeConfig: unknown) {
  if (runtimeConfig && typeof runtimeConfig === "object") {
    const value = (runtimeConfig as Record<string, unknown>).chatModel;
    if (typeof value === "string" && value.trim()) return value.trim();
  }
  return (
    process.env.COMPUTER_CHAT_MODEL_ID ||
    process.env.COMPUTER_CHAT_MODEL ||
    "us.anthropic.claude-haiku-4-5-20251001-v1:0"
  );
}

function buildComputerThreadSystemPrompt(input: {
  computer: Awaited<ReturnType<typeof loadComputer>>;
  threadTitle: string;
  requesterUserId: string | null;
  requesterContext?: Awaited<ReturnType<typeof assembleRequesterContext>>;
}) {
  const requesterOverlay = formatRequesterContextForPrompt(
    input.requesterContext,
  );
  return [
    `You are ${input.computer.name}, a shared ThinkWork Computer.`,
    "Answer for the requester of this turn, using only requester-scoped context and credentials when they are explicitly provided.",
    "Answer the user's Thread message directly. Be useful, concise, and grounded in the visible conversation.",
    `Requester user id: ${input.requesterUserId ?? "unavailable"}.`,
    `Thread: ${input.threadTitle}.`,
    `Workspace root: ${input.computer.live_workspace_root ?? "/workspace"}.`,
    requesterOverlay
      ? `<requester_context_overlay>\n${requesterOverlay}\n</requester_context_overlay>`
      : "",
  ]
    .filter(Boolean)
    .join("\n");
}

function threadTurnPayload(input: unknown) {
  const payload =
    input && typeof input === "object" && !Array.isArray(input)
      ? (input as Record<string, unknown>)
      : {};
  return {
    threadId: requiredPayloadString(payload.threadId, "threadId"),
    messageId: requiredPayloadString(payload.messageId, "messageId"),
    source:
      typeof payload.source === "string" && payload.source.trim()
        ? payload.source.trim()
        : "chat_message",
    actorType:
      typeof payload.actorType === "string" && payload.actorType.trim()
        ? payload.actorType.trim()
        : null,
    actorId:
      typeof payload.actorId === "string" && payload.actorId.trim()
        ? payload.actorId.trim()
        : null,
    requesterUserId:
      typeof payload.requesterUserId === "string" &&
      payload.requesterUserId.trim()
        ? payload.requesterUserId.trim()
        : null,
    contextClass:
      typeof payload.contextClass === "string" && payload.contextClass.trim()
        ? payload.contextClass.trim()
        : null,
    credentialSubject: credentialSubjectPayload(payload.credentialSubject),
    event: eventPayload(payload.event),
    surfaceContext: isRecord(payload.surfaceContext)
      ? payload.surfaceContext
      : {
          source:
            typeof payload.source === "string"
              ? payload.source
              : "chat_message",
        },
    runbookRunId:
      typeof payload.runbookRunId === "string" && payload.runbookRunId.trim()
        ? payload.runbookRunId.trim()
        : null,
  };
}

function requiredPayloadString(value: unknown, name: string) {
  if (typeof value !== "string" || !value.trim()) {
    throw new ComputerTaskDelegationError(
      `Computer task input missing ${name}`,
    );
  }
  return value.trim();
}

function credentialSubjectPayload(
  value: unknown,
): MemoryRequestContext["credentialSubject"] | undefined {
  if (!isRecord(value)) return undefined;
  const type =
    value.type === "service"
      ? "service"
      : value.type === "user"
        ? "user"
        : null;
  if (!type) return undefined;
  return {
    type,
    userId: optionalPayloadString(value.userId),
    connectionId: optionalPayloadString(value.connectionId),
    provider: optionalPayloadString(value.provider),
  };
}

function eventPayload(
  value: unknown,
): MemoryRequestContext["event"] | undefined {
  if (!isRecord(value)) return undefined;
  const metadata = isRecord(value.metadata) ? value.metadata : undefined;
  return {
    provider: optionalPayloadString(value.provider),
    eventType: optionalPayloadString(value.eventType),
    eventId: optionalPayloadString(value.eventId),
    metadata,
  };
}

function optionalPayloadString(value: unknown) {
  return typeof value === "string" && value.trim() ? value.trim() : null;
}
