import { createHash, randomBytes } from 'node:crypto'
import { createRequire } from 'node:module'
import {
  ControlPlaneQuotaExceededError,
  generateChannelInteractionToken,
  generateCloudApiToken,
  hashChannelInteractionToken,
  hashCloudApiToken,
  verifyCloudApiTokenHash,
  decodeSessionPageCursor,
  encodeSessionPageCursor,
} from './control-plane-store.ts'
import { redactAuditMetadata } from './audit-redaction.ts'
import type {
  WorkflowRunStatus,
  WorkflowStatus,
  CoordinationWatch,
} from '@open-cowork/shared'
import type {
  AttachWorkflowRunSessionInput,
  AppendWorkspaceEventInput,
  AuditEventRecord,
  AckChannelDeliveryInput,
  BindChannelSessionInput,
  ByokSecretRecord,
  ChannelProviderId,
  ClaimDueWorkflowRunInput,
  ClaimChannelDeliveryInput,
  ClaimChannelProviderEventInput,
  ClaimRateLimitInput,
  ClaimedWorkflowRunRecord,
  CloudWorkflowRecord,
  CloudWorkflowRunRecord,
  CommandQueueQuota,
  CompleteChannelProviderEventInput,
  ConsumeUsageQuotaInput,
  CompleteWorkflowRunInput,
  ControlPlaneRole,
  ControlPlaneSessionStatus,
  ControlPlaneStore,
  CreateChannelBindingInput,
  CreateChannelDeliveryInput,
  CreateChannelInteractionInput,
  CreateCloudCoordinationWatchInput,
  CreateAccountInput,
  CreateByokSecretInput,
  CreateHeadlessAgentInput,
  CreateWorkflowInput,
  CreateWorkflowRunInput,
  CreateThreadSmartFilterInput,
  CreateThreadTagInput,
  DisableByokSecretInput,
  EnqueueCommandInput,
  FailWorkflowRunInput,
  FindChannelInteractionInput,
  GrantApiTokenChannelBindingInput,
  IssuedApiTokenRecord,
  IssuedChannelInteractionRecord,
  IssuedManagedWorkerCredentialRecord,
  IssueApiTokenInput,
  ListApiTokenChannelBindingGrantsInput,
  ListCloudCoordinationWatchesInput,
  ListMatchingCloudCoordinationWatchesInput,
  ListSessionsPageInput,
  ManagedWorkerCredentialRecord,
  ManagedWorkerHeartbeatRecord,
  ManagedWorkerPoolRecord,
  ManagedWorkerPoolStatus,
  ManagedWorkerRecord,
  ManagedWorkerStatus,
  ListChannelDeliveriesInput,
  OrgMemberRecord,
  PrincipalMembershipRecord,
  RecordManagedWorkerHeartbeatInput,
  RecordAuditEventInput,
  RecordByokSecretValidationInput,
  RecordCloudAuthFailureInput,
  RecordUsageEventInput,
  ReapExpiredSessionLeasesInput,
  ReapedSessionLeaseRecord,
  ReapExpiredWorkflowClaimsInput,
  ReapedWorkflowClaimRecord,
  RevokeApiTokenInput,
  RevokeManagedWorkerCredentialInput,
  ResolvedManagedWorkerCredentialRecord,
  ResolveChannelInteractionInput,
  ResolveChannelInteractionWithCommandInput,
  SessionCommandRecord,
  SessionEventRecord,
  ThreadMetadataRecord,
  ThreadTagLinkInput,
  UsageEventRecord,
  UsageQuotaCounterRecord,
  UpdateChannelBindingInput,
  UpdateChannelCursorInput,
  UpdateCloudCoordinationWatchInput,
  UpdateHeadlessAgentInput,
  UpdateManagedWorkerPoolInput,
  UpdateManagedWorkerStatusInput,
  UpdateWorkflowStatusInput,
  UpdateThreadSmartFilterInput,
  UpdateThreadTagInput,
  UpsertBillingSubscriptionInput,
  UpsertChannelIdentityInput,
  UpsertMembershipInput,
  WorkerLeaseRecord,
  WorkerRole,
  WorkspaceEventCursorRecord,
  WorkspaceEventRecord,
  CreateManagedWorkerPoolInput,
  RegisterManagedWorkerInput,
  IssueManagedWorkerCredentialInput,
  ApiTokenChannelBindingGrantRecord,
} from './control-plane-store.ts'
import type {
  WorkflowWebhookReplayClaim,
  WorkflowWebhookSecurityStore,
} from '../workflow/workflow-webhook-server.ts'
import { runPostgresControlPlaneMigrations } from './postgres-migrations.ts'
import { workspaceEventCursorFromRow } from './workspace-event-cursor.ts'
import { normalizeChannelProviderId as normalizeProvider } from './channel-provider-utils.ts'
import { usageEventFromRow, billingSubscriptionFromRow } from './postgres-domains/billing.ts'
import { byokSecretFromRow } from './postgres-domains/byok.ts'
import { channelBindingFromRow, channelIdentityFromRow, channelInteractionFromRow, channelSessionBindingFromRow, headlessAgentFromRow } from './postgres-domains/channels.ts'
import {
  accountFromRow,
  apiTokenChannelBindingGrantFromRow,
  apiTokenFromRow,
  auditEventFromRow,
  cloudAuthBackoffFromRow,
  membershipFromRow,
  orgFromRow,
  tenantFromRow,
  userFromRow,
} from './postgres-domains/identity.ts'
import { migrationFromRow } from './postgres-domains/schema.ts'
import {
  commandFromRow,
  eventFromRow,
  heartbeatFromRow,
  leaseFromRow,
  projectionFromRow,
  sessionFromRow,
  settingFromRow,
  workspaceEventFromRow,
} from './postgres-domains/sessions.ts'
import { iso, jsonRecord, numberValue, type QueryResult, type QueryRow } from './postgres-domains/shared.ts'
import { threadSmartFilterFromRow, threadTagFromRow } from './postgres-domains/thread-index.ts'
import { webhookAuthFailureFromRow } from './postgres-domains/webhooks.ts'
import { workflowFromRow, workflowRunFromRow } from './postgres-domains/workflows.ts'
import { assertPostgresCommandEnqueueQuotas, assertPostgresCommandQueueQuota, assertPostgresConcurrentSessionQuota, assertPostgresWorkflowRunQuota, checkPostgresActiveWorkerQuota, listPostgresRunnableSessions } from './postgres-store-domains/quotas.ts'
import { PostgresManagedWorkersRepository } from './postgres-store-domains/workers.ts'
import { PostgresChannelProviderEventsRepository } from './postgres-store-domains/channel-provider-events.ts'
import { PostgresChannelDeliveriesRepository } from './postgres-store-domains/channel-deliveries.ts'
import { PostgresCoordinationWatchesRepository } from './postgres-store-domains/coordination-watches.ts'

type PgExecutor = {
  query<Row extends QueryRow = QueryRow>(text: string, values?: unknown[]): Promise<QueryResult<Row>>
}
type PgClient = PgExecutor & { release: () => void }
type PgPool = PgExecutor & { connect(): Promise<PgClient>; end(): Promise<void> }

export type PostgresControlPlaneStoreOptions = {
  connectionString: string
  runMigrations?: boolean
  pool?: PgPool
}

const require = createRequire(import.meta.url)

const THREAD_TAG_NAME_MAX_LENGTH = 48
const THREAD_SMART_FILTER_NAME_MAX_LENGTH = 64
const THREAD_DEFAULT_TAG_COLOR = '#64748b'
const THREAD_FILTER_MAX_VALUES = 50
const THREAD_BULK_MAX_SESSION_IDS = 500
const SMART_FILTER_QUERY_MAX_BYTES = 16_384
const WORKFLOW_RUN_LIST_LIMIT = 100
const CHANNEL_TEXT_MAX_LENGTH = 256
const CHANNEL_METADATA_MAX_BYTES = 16_384
const BYOK_PROVIDER_ID_MAX_LENGTH = 64
const BYOK_SECRET_TEXT_MAX_LENGTH = 4096
function loadPgPool(connectionString: string): PgPool {
  const pg = require('pg') as {
    Pool: new (options: { connectionString: string }) => PgPool
  }
  return new pg.Pool({ connectionString })
}

function nowIso(now: Date | undefined) {
  return (now || new Date()).toISOString()
}

function stableJson(value: unknown): string {
  if (Array.isArray(value)) return `[${value.map(stableJson).join(',')}]`
  if (value && typeof value === 'object') {
    return `{${Object.entries(value as Record<string, unknown>)
      .sort(([left], [right]) => left.localeCompare(right))
      .map(([field, entry]) => `${JSON.stringify(field)}:${stableJson(entry)}`)
      .join(',')}}`
  }
  return JSON.stringify(value)
}

function stableId(prefix: string, ...parts: string[]) {
  return `${prefix}_${createHash('sha256').update(parts.join('\0')).digest('hex').slice(0, 32)}`
}

function createWorkClaimToken(tenantId: string, workId: string, claimedBy: string) {
  return stableId('claim', tenantId, workId, claimedBy, randomBytes(16).toString('base64url'))
}

function workspaceOperationFromType(type: string) {
  if (/\b(created|submitted|uploaded|started)\b/.test(type)) return 'create'
  if (/\b(deleted|removed|archived)\b/.test(type)) return 'delete'
  return 'update'
}

function optionalTrimmedText(value: unknown) {
  return typeof value === 'string' && value.trim() ? value.trim() : null
}

function normalizeText(value: unknown, maxLength: number, label: string) {
  if (typeof value !== 'string' || !value.trim()) throw new Error(`${label} is required.`)
  const normalized = value.trim()
  if (normalized.length > maxLength) throw new Error(`${label} exceeds ${maxLength} characters.`)
  return normalized
}

function redactOperationalText(value: unknown, maxLength: number, label: string) {
  if (typeof value !== 'string' || !value.trim()) throw new Error(`${label} is required.`)
  const redacted = value.trim()
    .replace(/\b(Bearer\s+)[A-Za-z0-9._~+/-]+=*/gi, '$1[redacted]')
    .replace(/\b(api[_-]?key|token|secret|password|authorization)=([^\s&]+)/gi, '$1=[redacted]')
    .replace(/\b(gcp-sm|aws-sm|azure-kv|env):[^\s,)]+/gi, '$1:[redacted]')
    .replace(/\b(?:sk-[A-Za-z0-9._-]{6,}|oc[wc]_[A-Za-z0-9._-]{8,})\b/g, '[redacted]')
    .replace(/\b([A-Za-z0-9_-]{32,})\b/g, '[redacted]')
  return redacted.length <= maxLength ? redacted : `${redacted.slice(0, maxLength <= 3 ? maxLength : maxLength - 3)}${maxLength <= 3 ? '' : '...'}`
}

function normalizeOptionalText(value: unknown, maxLength: number, label: string) {
  if (value === undefined) return undefined
  return normalizeText(value, maxLength, label)
}

function normalizeTagColor(value: unknown) {
  return typeof value === 'string' && /^#[0-9a-fA-F]{6}$/.test(value.trim())
    ? value.trim()
    : THREAD_DEFAULT_TAG_COLOR
}

function normalizeIdList(values: readonly unknown[], label: string, maxLength: number) {
  if (!Array.isArray(values)) throw new Error(`${label} must be an array.`)
  if (values.length > maxLength) throw new Error(`${label} exceeds ${maxLength} entries.`)
  return [...new Set(values.map((value) => normalizeText(value, 256, label)))]
}

function normalizeThreadQuery(value: unknown) {
  const query = jsonRecord(value)
  const serialized = stableJson(query)
  if (Buffer.byteLength(serialized, 'utf8') > SMART_FILTER_QUERY_MAX_BYTES) {
    throw new Error(`Smart filter query exceeds ${SMART_FILTER_QUERY_MAX_BYTES} bytes.`)
  }
  return query
}

function normalizeRecord(value: unknown, label: string, maxBytes = CHANNEL_METADATA_MAX_BYTES): Record<string, unknown> {
  const record = jsonRecord(value)
  const serialized = stableJson(record)
  if (Buffer.byteLength(serialized, 'utf8') > maxBytes) {
    throw new Error(`${label} exceeds ${maxBytes} bytes.`)
  }
  return record
}

function normalizeNullableText(value: unknown, maxLength: number, label: string): string | null {
  if (value === undefined || value === null || value === '') return null
  return normalizeText(value, maxLength, label)
}

function normalizeNonNegativeInteger(value: unknown, label: string) {
  const parsed = Number(value ?? 0)
  if (!Number.isInteger(parsed) || parsed < 0) throw new Error(`${label} must be a non-negative integer.`)
  return parsed
}

function normalizePositiveInteger(value: unknown, label: string) {
  const parsed = Number(value ?? 0)
  if (!Number.isInteger(parsed) || parsed <= 0) throw new Error(`${label} must be a positive integer.`)
  return parsed
}

function windowStart(nowMs: number, windowMs: number) {
  return Math.floor(nowMs / windowMs) * windowMs
}

function retryAfterMs(nowMs: number, windowStartedAtMs: number, windowMs: number) {
  return Math.max(1, windowStartedAtMs + windowMs - nowMs)
}

function normalizeByokProviderId(value: unknown) {
  const providerId = normalizeText(value, BYOK_PROVIDER_ID_MAX_LENGTH, 'BYOK provider id').toLowerCase()
  if (!/^[a-z0-9][a-z0-9._-]*$/.test(providerId)) throw new Error(`Unsupported BYOK provider id ${providerId}.`)
  return providerId
}

export class PostgresControlPlaneStore implements ControlPlaneStore, WorkflowWebhookSecurityStore {
  private readonly pool: PgPool
  private readonly ownsPool: boolean
  private readonly managedWorkers: PostgresManagedWorkersRepository
  private readonly channelProviderEvents: PostgresChannelProviderEventsRepository
  private readonly channelDeliveries: PostgresChannelDeliveriesRepository
  private readonly coordinationWatches: PostgresCoordinationWatchesRepository
  private readonly quotaDeps = {
    lockQuota: (executor: PgExecutor, orgId: string, quotaKey: string, now?: Date) => this.lockQuota(executor, orgId, quotaKey, now),
    consumeUsageQuota: (executor: PgExecutor, input: ConsumeUsageQuotaInput) => this.consumeUsageQuotaWithExecutor(executor, input),
  }

  private constructor(pool: PgPool, ownsPool: boolean) {
    this.pool = pool
    this.ownsPool = ownsPool
    this.managedWorkers = new PostgresManagedWorkersRepository({
      pool: this.pool,
      withTransaction: (fn) => this.withTransaction(fn),
      recordAuditEvent: (executor, input) => this.recordAuditEventWithExecutor(executor, input),
    })
    this.channelProviderEvents = new PostgresChannelProviderEventsRepository({
      pool: this.pool,
      withTransaction: (fn) => this.withTransaction(fn),
    })
    this.channelDeliveries = new PostgresChannelDeliveriesRepository({
      pool: this.pool,
      withTransaction: (fn) => this.withTransaction(fn),
      consumeUsageQuota: (executor, input) => this.consumeUsageQuotaWithExecutor(executor, input),
    })
    this.coordinationWatches = new PostgresCoordinationWatchesRepository(this.pool)
  }

  static async connect(options: PostgresControlPlaneStoreOptions) {
    const pool = options.pool || loadPgPool(options.connectionString)
    const store = new PostgresControlPlaneStore(pool, !options.pool)
    if (options.runMigrations !== false) await store.runMigrations()
    return store
  }

  async close() {
    if (this.ownsPool) await this.pool.end()
  }

  async runMigrations() {
    await runPostgresControlPlaneMigrations(this.pool, (fn) => this.withTransaction(fn))
  }

  async createTenant(input: { tenantId: string, name: string, orgId?: string, createdAt?: Date }) {
    await this.pool.query(
      `INSERT INTO cloud_tenants (tenant_id, name, created_at)
       VALUES ($1, $2, $3)
       ON CONFLICT (tenant_id) DO NOTHING`,
      [input.tenantId, input.name, nowIso(input.createdAt)],
    )
    await this.ensureOrgForTenant({ tenantId: input.tenantId, name: input.name, orgId: input.orgId, createdAt: input.createdAt })
    return this.requireTenant(input.tenantId)
  }

  async ensureUser(input: {
    tenantId: string
    userId: string
    email: string
    role?: ControlPlaneRole
    createdAt?: Date
  }) {
    await this.requireTenant(input.tenantId)
    const existing = await this.maybeOne(
      `SELECT * FROM cloud_users WHERE tenant_id = $1 AND user_id = $2`,
      [input.tenantId, input.userId],
    )
    if (existing) return userFromRow(existing)
    await this.pool.query(
      `INSERT INTO cloud_users (tenant_id, user_id, email, role, created_at)
       VALUES ($1, $2, $3, $4, $5)
       ON CONFLICT (tenant_id, user_id) DO NOTHING`,
      [input.tenantId, input.userId, input.email, input.role || 'member', nowIso(input.createdAt)],
    )
    const org = await this.ensureOrgForTenant({ tenantId: input.tenantId, name: input.tenantId, createdAt: input.createdAt })
    const account = await this.createAccount({
      accountId: input.userId,
      idpSubject: input.userId,
      email: input.email,
      createdAt: input.createdAt,
    })
    await this.upsertMembership({
      orgId: org.orgId,
      accountId: account.accountId,
      role: input.role || 'member',
      status: 'active',
      updatedAt: input.createdAt,
      actor: { actorType: 'system', actorId: 'compat.ensureUser' },
    })
    return this.requireTenantUser(input.tenantId, input.userId)
  }

  async ensureOrgForTenant(input: { tenantId: string, name: string, orgId?: string, planKey?: string | null, status?: string, createdAt?: Date }) {
    const now = nowIso(input.createdAt)
    const result = await this.pool.query(
      `INSERT INTO cloud_orgs (org_id, tenant_id, name, plan_key, status, created_at, updated_at)
       VALUES ($1, $2, $3, $4, $5, $6, $6)
       ON CONFLICT (tenant_id) DO UPDATE
       SET name = COALESCE(NULLIF(EXCLUDED.name, ''), cloud_orgs.name),
           plan_key = COALESCE(EXCLUDED.plan_key, cloud_orgs.plan_key),
           status = cloud_orgs.status,
           updated_at = EXCLUDED.updated_at
       RETURNING *`,
      [input.orgId || input.tenantId, input.tenantId, input.name, input.planKey ?? null, input.status || 'active', now],
    )
    return orgFromRow(result.rows[0])
  }

  async createAccount(input: CreateAccountInput) {
    const now = nowIso(input.createdAt)
    const existing = await this.maybeOne(
      `SELECT * FROM cloud_accounts
       WHERE ($1::text IS NOT NULL AND idp_subject = $1)
          OR lower(email) = lower($2)
       ORDER BY updated_at DESC
       LIMIT 1`,
      [input.idpSubject || null, input.email],
    )
    if (existing) {
      const result = await this.pool.query(
        `UPDATE cloud_accounts
         SET idp_subject = COALESCE(cloud_accounts.idp_subject, $2),
             email = lower($3),
             display_name = COALESCE($4, cloud_accounts.display_name),
             updated_at = $5
         WHERE account_id = $1
         RETURNING *`,
        [
          existing.account_id,
          input.idpSubject || null,
          input.email,
          input.displayName || null,
          now,
        ],
      )
      return accountFromRow(result.rows[0])
    }
    const result = await this.pool.query(
      `INSERT INTO cloud_accounts (account_id, idp_subject, email, display_name, created_at, updated_at)
       VALUES ($1, $2, lower($3), $4, $5, $5)
       ON CONFLICT (account_id) DO UPDATE
       SET idp_subject = COALESCE(cloud_accounts.idp_subject, EXCLUDED.idp_subject),
           email = EXCLUDED.email,
           display_name = COALESCE(EXCLUDED.display_name, cloud_accounts.display_name),
           updated_at = EXCLUDED.updated_at
       RETURNING *`,
      [input.accountId, input.idpSubject || null, input.email, input.displayName || null, now],
    )
    return accountFromRow(result.rows[0])
  }

  async findAccountBySubject(idpSubject: string) {
    const row = await this.maybeOne(`SELECT * FROM cloud_accounts WHERE idp_subject = $1`, [idpSubject])
    return row ? accountFromRow(row) : null
  }

  async findAccountByEmail(email: string) {
    const row = await this.maybeOne(`SELECT * FROM cloud_accounts WHERE lower(email) = lower($1)`, [email])
    return row ? accountFromRow(row) : null
  }

  async upsertMembership(input: UpsertMembershipInput) {
    return this.withTransaction(async (client) => {
      const existing = await this.maybeOne(
        `SELECT * FROM cloud_memberships WHERE org_id = $1 AND account_id = $2`,
        [input.orgId, input.accountId],
        client,
      )
      const now = nowIso(input.updatedAt)
      const result = await client.query(
        `INSERT INTO cloud_memberships (org_id, account_id, role, status, created_at, updated_at)
         VALUES ($1, $2, $3, $4, $5, $5)
         ON CONFLICT (org_id, account_id) DO UPDATE
         SET role = EXCLUDED.role,
             status = EXCLUDED.status,
             updated_at = EXCLUDED.updated_at
         RETURNING *`,
        [input.orgId, input.accountId, input.role, input.status || 'active', now],
      )
      await this.recordAuditEventWithExecutor(client, {
        orgId: input.orgId,
        accountId: input.accountId,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: existing ? 'membership.updated' : 'membership.created',
        targetType: 'membership',
        targetId: `${input.orgId}:${input.accountId}`,
        metadata: { role: input.role, status: input.status || 'active' },
        createdAt: input.updatedAt,
      })
      return membershipFromRow(result.rows[0])
    })
  }

  async listOrgMembers(orgId: string, input: { query?: string | null, limit?: number | null } = {}) {
    const queryText = input.query?.trim() || null
    const result = await this.pool.query(
      `SELECT
         m.org_id,
         m.account_id,
         a.email,
         a.display_name,
         m.role,
         m.status,
         m.created_at,
         m.updated_at
       FROM cloud_memberships m
       JOIN cloud_accounts a ON a.account_id = m.account_id
       WHERE m.org_id = $1
         AND (
           $2::text IS NULL
           OR m.account_id ILIKE '%' || $2 || '%'
           OR a.email ILIKE '%' || $2 || '%'
           OR COALESCE(a.display_name, '') ILIKE '%' || $2 || '%'
           OR m.role ILIKE '%' || $2 || '%'
           OR m.status ILIKE '%' || $2 || '%'
         )
       ORDER BY m.updated_at DESC, a.email ASC
       LIMIT $3`,
      [orgId, queryText, Math.max(1, Math.min(input.limit || 100, 500))],
    )
    return result.rows.map((row): OrgMemberRecord => ({
      orgId: String(row.org_id),
      accountId: String(row.account_id),
      email: String(row.email),
      displayName: row.display_name ? String(row.display_name) : null,
      role: row.role as OrgMemberRecord['role'],
      status: row.status as OrgMemberRecord['status'],
      createdAt: iso(row.created_at),
      updatedAt: iso(row.updated_at),
    }))
  }

  async listMembershipsForAccount(accountId: string) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_memberships WHERE account_id = $1 ORDER BY updated_at DESC, org_id`,
      [accountId],
    )
    return result.rows.map(membershipFromRow)
  }

  async resolvePrincipalMembership(input: { tenantId: string, userId?: string | null, accountId?: string | null, idpSubject?: string | null, email?: string | null }): Promise<PrincipalMembershipRecord | null> {
    const row = await this.maybeOne(
      `SELECT
         o.org_id, o.tenant_id, o.name AS org_name, o.plan_key, o.status AS org_status,
         o.created_at AS org_created_at, o.updated_at AS org_updated_at,
         a.account_id, a.idp_subject, a.email, a.display_name,
         a.created_at AS account_created_at, a.updated_at AS account_updated_at,
         m.role, m.status AS membership_status,
         m.created_at AS membership_created_at, m.updated_at AS membership_updated_at
       FROM cloud_orgs o
       JOIN cloud_memberships m ON m.org_id = o.org_id
       JOIN cloud_accounts a ON a.account_id = m.account_id
       WHERE (o.tenant_id = $1 OR o.org_id = $1)
         AND (
           ($2::text IS NOT NULL AND a.account_id = $2)
           OR ($3::text IS NOT NULL AND a.idp_subject = $3)
           OR ($4::text IS NOT NULL AND lower(a.email) = lower($4))
           OR ($5::text IS NOT NULL AND a.account_id = $5)
         )
       ORDER BY
         CASE
           WHEN $2::text IS NOT NULL AND a.account_id = $2 THEN 0
           WHEN $3::text IS NOT NULL AND a.idp_subject = $3 THEN 1
           WHEN $4::text IS NOT NULL AND lower(a.email) = lower($4) THEN 2
           WHEN $5::text IS NOT NULL AND a.account_id = $5 THEN 3
           ELSE 4
         END,
         m.updated_at DESC
       LIMIT 1`,
      [input.tenantId, input.accountId || null, input.idpSubject || null, input.email || null, input.userId || null],
    )
    if (!row) return null
    return {
      org: orgFromRow({
        org_id: row.org_id,
        tenant_id: row.tenant_id,
        name: row.org_name,
        plan_key: row.plan_key,
        status: row.org_status,
        created_at: row.org_created_at,
        updated_at: row.org_updated_at,
      }),
      account: accountFromRow({
        account_id: row.account_id,
        idp_subject: row.idp_subject,
        email: row.email,
        display_name: row.display_name,
        created_at: row.account_created_at,
        updated_at: row.account_updated_at,
      }),
      membership: membershipFromRow({
        org_id: row.org_id,
        account_id: row.account_id,
        role: row.role,
        status: row.membership_status,
        created_at: row.membership_created_at,
        updated_at: row.membership_updated_at,
      }),
    }
  }

  async issueApiToken(input: IssueApiTokenInput): Promise<IssuedApiTokenRecord> {
    return this.withTransaction(async (client) => {
      const generated = generateCloudApiToken(input)
      const now = nowIso(input.createdAt)
      const result = await client.query(
        `INSERT INTO cloud_api_tokens (
          token_id, org_id, account_id, name, token_hash, scopes, last4,
          expires_at, revoked_at, last_used_at, created_at, updated_at
         )
         VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, NULL, NULL, $9, $9)
         RETURNING *`,
        [
          generated.tokenId,
          input.orgId,
          input.accountId || null,
          normalizeText(input.name, 96, 'API token name'),
          hashCloudApiToken(generated.plaintext),
          JSON.stringify([...new Set(input.scopes)]),
          generated.plaintext.slice(-4),
          input.expiresAt ? input.expiresAt.toISOString() : null,
          now,
        ],
      )
      const token = apiTokenFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: input.orgId,
        accountId: input.accountId || null,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: 'api_token.created',
        targetType: 'api_token',
        targetId: token.tokenId,
        metadata: { name: token.name, scopes: token.scopes, last4: token.last4 },
        createdAt: input.createdAt,
      })
      return { token, plaintext: generated.plaintext }
    })
  }

  async findApiTokenByPlaintext(plaintext: string, now = new Date()) {
    const nowText = nowIso(now)
    const candidates = await this.pool.query(
      `SELECT *
       FROM cloud_api_tokens
       WHERE left($1, length('occ_' || token_id || '_')) = ('occ_' || token_id || '_')
         AND revoked_at IS NULL
         AND (expires_at IS NULL OR expires_at > $2)
       ORDER BY created_at DESC`,
      [plaintext, nowText],
    )
    const matched = candidates.rows.find((row) => verifyCloudApiTokenHash(plaintext, String(row.token_hash)))
    if (!matched) return null
    const result = await this.pool.query(
      `UPDATE cloud_api_tokens
       SET last_used_at = $2, updated_at = $2
       WHERE token_id = $1
         AND revoked_at IS NULL
         AND (expires_at IS NULL OR expires_at > $2)
       RETURNING *`,
      [String(matched.token_id), nowText],
    )
    return result.rows[0] ? apiTokenFromRow(result.rows[0]) : null
  }

  async listApiTokens(orgId: string) {
    const result = await this.pool.query(
      `SELECT *
       FROM cloud_api_tokens
       WHERE org_id = $1
       ORDER BY created_at DESC`,
      [orgId],
    )
    return result.rows.map(apiTokenFromRow)
  }

  async revokeApiToken(input: RevokeApiTokenInput) {
    return this.withTransaction(async (client) => {
      const now = nowIso(input.revokedAt)
      const result = await client.query(
        `UPDATE cloud_api_tokens
         SET revoked_at = COALESCE(revoked_at, $2), updated_at = $2
         WHERE token_id = $1
           AND ($3::text IS NULL OR org_id = $3)
         RETURNING *`,
        [input.tokenId, now, input.orgId || null],
      )
      if (!result.rows[0]) return null
      const token = apiTokenFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: token.orgId,
        accountId: token.accountId,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: 'api_token.revoked',
        targetType: 'api_token',
        targetId: token.tokenId,
        metadata: { name: token.name, scopes: token.scopes, last4: token.last4 },
        createdAt: input.revokedAt,
      })
      return token
    })
  }

  async grantApiTokenChannelBinding(input: GrantApiTokenChannelBindingInput): Promise<ApiTokenChannelBindingGrantRecord> {
    return this.withTransaction(async (client) => {
      const tokenRow = await this.maybeOne(
        `SELECT * FROM cloud_api_tokens WHERE org_id = $1 AND token_id = $2`,
        [input.orgId, input.tokenId],
        client,
      )
      if (!tokenRow) throw new Error(`Unknown API token ${input.tokenId}.`)
      const bindingRow = await this.maybeOne(
        `SELECT * FROM cloud_channel_bindings WHERE org_id = $1 AND binding_id = $2`,
        [input.orgId, input.channelBindingId],
        client,
      )
      if (!bindingRow) throw new Error(`Unknown channel binding ${input.channelBindingId}.`)
      const createdAt = nowIso(input.createdAt)
      const result = await client.query(
        `INSERT INTO cloud_api_token_channel_binding_grants (
          org_id, token_id, channel_binding_id, created_at
         )
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (org_id, token_id, channel_binding_id) DO NOTHING
         RETURNING *`,
        [input.orgId, input.tokenId, input.channelBindingId, createdAt],
      )
      const row = result.rows[0] || await this.one(
        `SELECT * FROM cloud_api_token_channel_binding_grants
         WHERE org_id = $1 AND token_id = $2 AND channel_binding_id = $3`,
        [input.orgId, input.tokenId, input.channelBindingId],
        client,
      )
      if (result.rows[0]) {
        const token = apiTokenFromRow(tokenRow)
        await this.recordAuditEventWithExecutor(client, {
          orgId: input.orgId,
          accountId: input.actor?.accountId || token.accountId,
          actorType: input.actor?.actorType || 'system',
          actorId: input.actor?.actorId || null,
          eventType: 'api_token.channel_binding_granted',
          targetType: 'api_token',
          targetId: input.tokenId,
          metadata: { channelBindingId: input.channelBindingId },
          createdAt: input.createdAt,
        })
      }
      return apiTokenChannelBindingGrantFromRow(row)
    })
  }

  async listApiTokenChannelBindingGrants(input: ListApiTokenChannelBindingGrantsInput): Promise<ApiTokenChannelBindingGrantRecord[]> {
    const result = await this.pool.query(
      `SELECT * FROM cloud_api_token_channel_binding_grants
       WHERE org_id = $1 AND token_id = $2
       ORDER BY channel_binding_id`,
      [input.orgId, input.tokenId],
    )
    return result.rows.map(apiTokenChannelBindingGrantFromRow)
  }

  async createManagedWorkerPool(input: CreateManagedWorkerPoolInput): Promise<ManagedWorkerPoolRecord> {
    return this.managedWorkers.createPool(input)
  }

  async updateManagedWorkerPool(input: UpdateManagedWorkerPoolInput): Promise<ManagedWorkerPoolRecord | null> {
    return this.managedWorkers.updatePool(input)
  }

  async getManagedWorkerPool(orgId: string, poolId: string) {
    return this.managedWorkers.getPool(orgId, poolId)
  }

  async listManagedWorkerPools(orgId: string, input: { status?: ManagedWorkerPoolStatus | null, limit?: number | null } = {}) {
    return this.managedWorkers.listPools(orgId, input)
  }

  async registerManagedWorker(input: RegisterManagedWorkerInput): Promise<ManagedWorkerRecord> {
    return this.managedWorkers.registerWorker(input)
  }

  async updateManagedWorkerStatus(input: UpdateManagedWorkerStatusInput): Promise<ManagedWorkerRecord | null> {
    return this.managedWorkers.updateWorkerStatus(input)
  }

  async getManagedWorker(orgId: string, workerId: string) {
    return this.managedWorkers.getWorker(orgId, workerId)
  }

  async listManagedWorkers(orgId: string, input: { poolId?: string | null, status?: ManagedWorkerStatus | null, limit?: number | null } = {}) {
    return this.managedWorkers.listWorkers(orgId, input)
  }

  async issueManagedWorkerCredential(input: IssueManagedWorkerCredentialInput): Promise<IssuedManagedWorkerCredentialRecord> {
    return this.managedWorkers.issueCredential(input)
  }

  async listManagedWorkerCredentials(orgId: string, workerId: string) {
    return this.managedWorkers.listCredentials(orgId, workerId)
  }

  async findManagedWorkerCredentialByPlaintext(plaintext: string, now = new Date()): Promise<ResolvedManagedWorkerCredentialRecord | null> {
    return this.managedWorkers.findCredentialByPlaintext(plaintext, now)
  }

  async revokeManagedWorkerCredential(input: RevokeManagedWorkerCredentialInput): Promise<ManagedWorkerCredentialRecord | null> {
    return this.managedWorkers.revokeCredential(input)
  }

  async recordManagedWorkerHeartbeat(input: RecordManagedWorkerHeartbeatInput): Promise<ManagedWorkerHeartbeatRecord> {
    return this.managedWorkers.recordHeartbeat(input)
  }

  async listManagedWorkerHeartbeats(orgId: string, input: { workerId?: string | null, limit?: number | null } = {}) {
    return this.managedWorkers.listHeartbeats(orgId, input)
  }

  async recordAuditEvent(input: RecordAuditEventInput) {
    return this.recordAuditEventWithExecutor(this.pool, input)
  }

  async listAuditEvents(orgId: string, limit = 100) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_audit_events
       WHERE org_id = $1
       ORDER BY created_at DESC, event_id DESC
       LIMIT $2`,
      [orgId, Math.max(1, Math.min(limit, 500))],
    )
    return result.rows.map(auditEventFromRow)
  }

  async consumeUsageQuota(input: ConsumeUsageQuotaInput) {
    return this.withTransaction((client) => this.consumeUsageQuotaWithExecutor(client, input))
  }

  async listUsageQuotaCounters(orgId: string): Promise<UsageQuotaCounterRecord[]> {
    const result = await this.pool.query(
      `SELECT org_id, quota_key, window_started_at_ms, quantity
       FROM cloud_usage_counters
       WHERE org_id = $1
       ORDER BY quota_key ASC`,
      [orgId],
    )
    return result.rows.map((row) => ({
      orgId: String(row.org_id),
      quotaKey: String(row.quota_key),
      windowStartedAtMs: numberValue(row.window_started_at_ms),
      quantity: numberValue(row.quantity),
    }))
  }

  async recordUsageEvent(input: RecordUsageEventInput) {
    return this.recordUsageEventWithExecutor(this.pool, input)
  }

  async listUsageEvents(orgId: string, limit = 100) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_usage_events
       WHERE org_id = $1
       ORDER BY created_at DESC, event_id DESC
       LIMIT $2`,
      [orgId, Math.max(1, Math.min(limit, 500))],
    )
    return result.rows.map(usageEventFromRow)
  }

  async upsertBillingSubscription(input: UpsertBillingSubscriptionInput) {
    await this.requireOrg(input.orgId)
    const now = nowIso(input.updatedAt)
    const priorRow = await this.maybeOne(
      `SELECT * FROM cloud_subscriptions WHERE org_id = $1`,
      [input.orgId],
    )
    const prior = priorRow ? billingSubscriptionFromRow(priorRow) : null
    const result = await this.pool.query(
      `INSERT INTO cloud_subscriptions (
        org_id, plan_key, provider_id, provider_customer_id, provider_subscription_id,
        status, seats, entitlements, current_period_end, cancel_at_period_end,
        metadata, created_at, updated_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9, $10, $11::jsonb, $12, $12)
       ON CONFLICT (org_id) DO UPDATE
       SET plan_key = EXCLUDED.plan_key,
           provider_id = EXCLUDED.provider_id,
           provider_customer_id = EXCLUDED.provider_customer_id,
           provider_subscription_id = EXCLUDED.provider_subscription_id,
           status = EXCLUDED.status,
           seats = EXCLUDED.seats,
           entitlements = EXCLUDED.entitlements,
           current_period_end = EXCLUDED.current_period_end,
           cancel_at_period_end = EXCLUDED.cancel_at_period_end,
           metadata = EXCLUDED.metadata,
           updated_at = EXCLUDED.updated_at
       RETURNING *`,
      [
        input.orgId,
        normalizeText(input.planKey, CHANNEL_TEXT_MAX_LENGTH, 'Billing plan key'),
        normalizeText(input.providerId, CHANNEL_TEXT_MAX_LENGTH, 'Billing provider id'),
        input.providerCustomerId ? normalizeText(input.providerCustomerId, CHANNEL_TEXT_MAX_LENGTH, 'Billing provider customer id') : null,
        input.providerSubscriptionId ? normalizeText(input.providerSubscriptionId, CHANNEL_TEXT_MAX_LENGTH, 'Billing provider subscription id') : null,
        input.status,
        normalizePositiveInteger(input.seats || 1, 'Billing seats'),
        JSON.stringify(jsonRecord(input.entitlements)),
        input.currentPeriodEnd ? (input.currentPeriodEnd instanceof Date ? input.currentPeriodEnd.toISOString() : new Date(input.currentPeriodEnd).toISOString()) : null,
        input.cancelAtPeriodEnd === true,
        JSON.stringify(redactAuditMetadata(input.metadata)),
        now,
      ],
    )
    const subscription = billingSubscriptionFromRow(result.rows[0])
    await this.recordAuditEvent({
      orgId: subscription.orgId,
      actorType: 'system',
      actorId: 'billing.subscription.upsert',
      eventType: prior ? 'billing.subscription.updated' : 'billing.subscription.created',
      targetType: 'billing_subscription',
      targetId: subscription.providerSubscriptionId || subscription.orgId,
      metadata: {
        providerId: subscription.providerId,
        previousPlanKey: prior?.planKey || null,
        previousStatus: prior?.status || null,
        previousEntitlementsHash: prior ? stableJson(prior.entitlements) : null,
        planKey: subscription.planKey,
        status: subscription.status,
        entitlementsHash: stableJson(subscription.entitlements),
        seats: subscription.seats,
        providerCustomerId: subscription.providerCustomerId,
        providerSubscriptionId: subscription.providerSubscriptionId,
        providerEventId: input.metadata?.stripeEventId || input.metadata?.eventId || null,
      },
      createdAt: input.updatedAt,
    })
    return subscription
  }

  async getBillingSubscription(orgId: string) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_subscriptions WHERE org_id = $1`,
      [orgId],
    )
    return row ? billingSubscriptionFromRow(row) : null
  }

  async findBillingSubscriptionByProvider(input: {
    providerId: string
    providerCustomerId?: string | null
    providerSubscriptionId?: string | null
  }) {
    const providerId = normalizeText(input.providerId, CHANNEL_TEXT_MAX_LENGTH, 'Billing provider id')
    const row = input.providerSubscriptionId
      ? await this.maybeOne(
        `SELECT * FROM cloud_subscriptions
         WHERE provider_id = $1 AND provider_subscription_id = $2`,
        [providerId, input.providerSubscriptionId],
      )
      : input.providerCustomerId
        ? await this.maybeOne(
          `SELECT * FROM cloud_subscriptions
           WHERE provider_id = $1 AND provider_customer_id = $2
           ORDER BY updated_at DESC
           LIMIT 1`,
          [providerId, input.providerCustomerId],
        )
        : null
    return row ? billingSubscriptionFromRow(row) : null
  }

  async claimRateLimit(input: ClaimRateLimitInput) {
    const limit = normalizePositiveInteger(input.limit, 'Rate limit')
    const windowMs = normalizePositiveInteger(input.windowMs, 'Rate-limit window')
    const now = input.now || new Date()
    const nowMs = now.getTime()
    const startedAtMs = windowStart(nowMs, windowMs)
    const result = await this.pool.query(
      `INSERT INTO cloud_rate_limits (scope, source, window_started_at_ms, count)
       VALUES ($1, $2, $3, 1)
       ON CONFLICT (scope, source) DO UPDATE
       SET window_started_at_ms = CASE
             WHEN cloud_rate_limits.window_started_at_ms = $3 THEN cloud_rate_limits.window_started_at_ms
             ELSE $3
           END,
           count = CASE
             WHEN cloud_rate_limits.window_started_at_ms = $3 THEN cloud_rate_limits.count + 1
             ELSE 1
           END
       RETURNING count, window_started_at_ms`,
      [input.scope, input.source, startedAtMs],
    )
    const count = numberValue(result.rows[0]?.count)
    const resetMs = retryAfterMs(nowMs, numberValue(result.rows[0]?.window_started_at_ms), windowMs)
    return {
      allowed: count <= limit,
      scope: input.scope,
      source: input.source,
      limit,
      count,
      resetAt: new Date(nowMs + resetMs).toISOString(),
      retryAfterMs: resetMs,
      policyCode: input.policyCode,
    }
  }

  async checkCloudAuthBackoff(input: { scope: string, source?: string, now?: Date }) {
    const nowMs = (input.now || new Date()).getTime()
    const row = await this.maybeOne(
      `SELECT * FROM cloud_auth_failures WHERE scope = $1`,
      [input.scope],
    )
    if (!row) {
      return {
        allowed: true,
        scope: input.scope,
        source: input.source || input.scope,
        failureCount: 0,
        blockedUntilMs: 0,
        retryAfterMs: 0,
      }
    }
    return cloudAuthBackoffFromRow(row, nowMs)
  }

  async recordCloudAuthFailure(input: RecordCloudAuthFailureInput) {
    const nowMs = (input.now || new Date()).getTime()
    const windowMs = normalizePositiveInteger(input.windowMs, 'Auth backoff window')
    const limit = normalizePositiveInteger(input.limit, 'Auth failure limit')
    const backoffMs = normalizePositiveInteger(input.backoffMs, 'Auth backoff duration')
    const startedAtMs = windowStart(nowMs, windowMs)
    const blockedUntilMs = nowMs + backoffMs
    const result = await this.pool.query(
      `INSERT INTO cloud_auth_failures (
        scope, source, auth_window_started_at_ms, auth_failure_count, blocked_until_ms
       )
       VALUES ($1, $2, $3, 1, CASE WHEN $4 <= 1 THEN $5 ELSE 0 END)
       ON CONFLICT (scope) DO UPDATE
       SET source = EXCLUDED.source,
           auth_window_started_at_ms = CASE
             WHEN cloud_auth_failures.auth_window_started_at_ms = $3 THEN cloud_auth_failures.auth_window_started_at_ms
             ELSE $3
           END,
           auth_failure_count = CASE
             WHEN cloud_auth_failures.auth_window_started_at_ms = $3 THEN cloud_auth_failures.auth_failure_count + 1
             ELSE 1
           END,
           blocked_until_ms = CASE
             WHEN (
               CASE
                 WHEN cloud_auth_failures.auth_window_started_at_ms = $3 THEN cloud_auth_failures.auth_failure_count + 1
                 ELSE 1
               END
             ) >= $4 THEN GREATEST(cloud_auth_failures.blocked_until_ms, $5)
             ELSE cloud_auth_failures.blocked_until_ms
           END
       RETURNING *`,
      [input.scope, input.source, startedAtMs, limit, blockedUntilMs],
    )
    return cloudAuthBackoffFromRow(result.rows[0], nowMs)
  }

  async createByokSecret(input: CreateByokSecretInput) {
    return this.withTransaction(async (client) => {
      const providerId = normalizeByokProviderId(input.providerId)
      const ciphertext = normalizeNullableText(input.ciphertext, BYOK_SECRET_TEXT_MAX_LENGTH, 'BYOK ciphertext')
      const kmsRef = normalizeNullableText(input.kmsRef, BYOK_SECRET_TEXT_MAX_LENGTH, 'BYOK KMS ref')
      if ((ciphertext && kmsRef) || (!ciphertext && !kmsRef)) {
        throw new Error('BYOK secret requires exactly one of ciphertext or kmsRef.')
      }
      const status = input.status || 'pending_validation'
      const now = nowIso(input.createdAt)
      const prior = status === 'active'
        ? await client.query(
          `UPDATE cloud_byok_secrets
           SET status = 'disabled', updated_at = $3
           WHERE org_id = $1 AND provider_id = $2 AND status = 'active'
           RETURNING *`,
          [input.orgId, providerId, now],
        )
        : await client.query(
          `SELECT * FROM cloud_byok_secrets
           WHERE org_id = $1 AND provider_id = $2 AND status = 'active'
           LIMIT 1`,
          [input.orgId, providerId],
        )
      const priorActive = prior.rows[0] ? byokSecretFromRow(prior.rows[0]) : null
      const result = await client.query(
        `INSERT INTO cloud_byok_secrets (
          secret_id, org_id, provider_id, status, ciphertext, kms_ref, last4,
          key_fingerprint, created_by_account_id, rotated_from_secret_id,
          last_validated_at, validation_error, created_at, updated_at
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NULL, NULL, $11, $11)
        RETURNING *`,
        [
          normalizeText(input.secretId, CHANNEL_TEXT_MAX_LENGTH, 'BYOK secret id'),
          input.orgId,
          providerId,
          status,
          ciphertext,
          kmsRef,
          normalizeText(input.last4, 32, 'BYOK secret last4'),
          normalizeText(input.keyFingerprint, 128, 'BYOK key fingerprint'),
          input.createdByAccountId || null,
          input.rotatedFromSecretId || priorActive?.secretId || null,
          now,
        ],
      )
      const secret = byokSecretFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: secret.orgId,
        accountId: secret.createdByAccountId,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: priorActive
          ? status === 'active'
            ? 'byok_secret.rotated'
            : 'byok_secret.rotation_started'
          : 'byok_secret.created',
        targetType: 'byok_secret',
        targetId: secret.secretId,
        metadata: {
          providerId: secret.providerId,
          status: secret.status,
          last4: secret.last4,
          keyFingerprint: secret.keyFingerprint,
          rotatedFromSecretId: secret.rotatedFromSecretId,
        },
        createdAt: input.createdAt,
      })
      return secret
    })
  }

  async getByokSecret(orgId: string, providerId: string) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_byok_secrets
       WHERE org_id = $1 AND provider_id = $2
       ORDER BY updated_at DESC, created_at DESC, secret_id DESC
       LIMIT 1`,
      [orgId, normalizeByokProviderId(providerId)],
    )
    return row ? byokSecretFromRow(row) : null
  }

  async getActiveByokSecret(orgId: string, providerId: string) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_byok_secrets
       WHERE org_id = $1 AND provider_id = $2 AND status = 'active'
       ORDER BY updated_at DESC, created_at DESC, secret_id DESC
       LIMIT 1`,
      [orgId, normalizeByokProviderId(providerId)],
    )
    return row ? byokSecretFromRow(row) : null
  }

  async listByokSecrets(orgId: string) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_byok_secrets
       WHERE org_id = $1
       ORDER BY updated_at DESC, created_at DESC, provider_id, secret_id DESC`,
      [orgId],
    )
    return result.rows.map(byokSecretFromRow)
  }

  async disableByokSecret(input: DisableByokSecretInput) {
    return this.withTransaction(async (client) => {
      const providerId = normalizeByokProviderId(input.providerId)
      const now = nowIso(input.disabledAt)
      const result = await client.query(
        `UPDATE cloud_byok_secrets
         SET status = 'disabled', updated_at = $4
         WHERE org_id = $1
           AND provider_id = $2
           AND ($3::text IS NULL OR secret_id = $3)
           AND status <> 'disabled'
         RETURNING *`,
        [input.orgId, providerId, input.secretId || null, now],
      )
      if (!result.rows[0]) return null
      const secrets = result.rows
        .map(byokSecretFromRow)
        .sort((left, right) => (
          right.updatedAt.localeCompare(left.updatedAt)
          || right.createdAt.localeCompare(left.createdAt)
          || right.secretId.localeCompare(left.secretId)
        ))
      for (const secret of secrets) {
        await this.recordAuditEventWithExecutor(client, {
          orgId: secret.orgId,
          accountId: input.actor?.accountId || secret.createdByAccountId,
          actorType: input.actor?.actorType || 'system',
          actorId: input.actor?.actorId || null,
          eventType: 'byok_secret.disabled',
          targetType: 'byok_secret',
          targetId: secret.secretId,
          metadata: { providerId: secret.providerId, status: secret.status, last4: secret.last4, keyFingerprint: secret.keyFingerprint },
          createdAt: input.disabledAt,
        })
      }
      return secrets[0]
    })
  }

  async recordByokSecretValidation(input: RecordByokSecretValidationInput) {
    return this.withTransaction(async (client) => {
      const providerId = normalizeByokProviderId(input.providerId)
      const now = nowIso(input.validatedAt)
      let priorActive: ByokSecretRecord | null = null
      let targetSecretId = input.secretId || null
      if (input.status === 'active') {
        const target = await client.query(
          `SELECT * FROM cloud_byok_secrets
           WHERE org_id = $1
             AND provider_id = $2
             AND ($3::text IS NULL OR secret_id = $3)
             AND ($3::text IS NOT NULL OR status = 'active')
           ORDER BY updated_at DESC, created_at DESC, secret_id DESC
           LIMIT 1
           FOR UPDATE`,
          [input.orgId, providerId, targetSecretId],
        )
        if (!target.rows[0]) return null
        targetSecretId = String(target.rows[0].secret_id)
        const prior = await client.query(
          `UPDATE cloud_byok_secrets
           SET status = 'disabled', updated_at = $4
           WHERE org_id = $1
             AND provider_id = $2
             AND secret_id <> $3
             AND status = 'active'
           RETURNING *`,
          [input.orgId, providerId, targetSecretId, now],
        )
        priorActive = prior.rows
          .map(byokSecretFromRow)
          .sort((left, right) => (
            right.updatedAt.localeCompare(left.updatedAt)
            || right.createdAt.localeCompare(left.createdAt)
            || right.secretId.localeCompare(left.secretId)
          ))[0] || null
      }
      const result = await client.query(
        `UPDATE cloud_byok_secrets
         SET status = COALESCE($4, status),
             last_validated_at = $5,
             validation_error = $6,
             rotated_from_secret_id = COALESCE(rotated_from_secret_id, $7),
             updated_at = $5
         WHERE org_id = $1
           AND provider_id = $2
           AND ($3::text IS NULL OR secret_id = $3)
           AND ($3::text IS NOT NULL OR status = 'active')
         RETURNING *`,
        [
          input.orgId,
          providerId,
          targetSecretId,
          input.status || null,
          now,
          input.validationError || null,
          input.status === 'active' ? priorActive?.secretId || null : null,
        ],
      )
      if (!result.rows[0]) return null
      const secret = byokSecretFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: secret.orgId,
        accountId: input.actor?.accountId || secret.createdByAccountId,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: 'byok_secret.validated',
        targetType: 'byok_secret',
        targetId: secret.secretId,
        metadata: {
          providerId: secret.providerId,
          status: secret.status,
          last4: secret.last4,
          keyFingerprint: secret.keyFingerprint,
          validationError: secret.validationError,
        },
        createdAt: input.validatedAt,
      })
      if (input.status === 'active' && (priorActive || secret.rotatedFromSecretId)) {
        await this.recordAuditEventWithExecutor(client, {
          orgId: secret.orgId,
          accountId: input.actor?.accountId || secret.createdByAccountId,
          actorType: input.actor?.actorType || 'system',
          actorId: input.actor?.actorId || null,
          eventType: 'byok_secret.rotated',
          targetType: 'byok_secret',
          targetId: secret.secretId,
          metadata: {
            providerId: secret.providerId,
            status: secret.status,
            last4: secret.last4,
            keyFingerprint: secret.keyFingerprint,
            rotatedFromSecretId: secret.rotatedFromSecretId || priorActive?.secretId || null,
          },
          createdAt: input.validatedAt,
        })
      }
      return secret
    })
  }

  async createHeadlessAgent(input: CreateHeadlessAgentInput) {
    return this.withTransaction(async (client) => {
      const now = nowIso(input.createdAt)
      const result = await client.query(
        `INSERT INTO headless_agents (
          agent_id, org_id, tenant_id, profile_name, name, status, managed,
          created_by_account_id, created_at, updated_at
         )
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9)
         ON CONFLICT (agent_id) DO NOTHING
         RETURNING *`,
        [
          normalizeText(input.agentId, CHANNEL_TEXT_MAX_LENGTH, 'Headless agent id'),
          input.orgId,
          input.tenantId,
          normalizeText(input.profileName, CHANNEL_TEXT_MAX_LENGTH, 'Headless agent profile'),
          normalizeText(input.name, CHANNEL_TEXT_MAX_LENGTH, 'Headless agent name'),
          input.status || 'active',
          input.managed === true,
          input.createdByAccountId || null,
          now,
        ],
      )
      const row = result.rows[0] || await this.one(
        `SELECT * FROM headless_agents WHERE agent_id = $1 AND org_id = $2`,
        [input.agentId, input.orgId],
        client,
      )
      const agent = headlessAgentFromRow(row)
      if (result.rows[0]) {
        await this.recordAuditEventWithExecutor(client, {
          orgId: agent.orgId,
          accountId: agent.createdByAccountId,
          actorType: 'system',
          actorId: 'headless_agent.create',
          eventType: 'headless_agent.created',
          targetType: 'headless_agent',
          targetId: agent.agentId,
          metadata: { name: agent.name, profileName: agent.profileName, managed: agent.managed },
          createdAt: input.createdAt,
        })
      }
      return agent
    })
  }

  async updateHeadlessAgent(input: UpdateHeadlessAgentInput) {
    return this.withTransaction(async (client) => {
      const result = await client.query(
        `UPDATE headless_agents
         SET profile_name = CASE WHEN $3::boolean THEN $4 ELSE profile_name END,
             name = CASE WHEN $5::boolean THEN $6 ELSE name END,
             status = COALESCE($7, status),
             managed = CASE WHEN $8::boolean THEN $9 ELSE managed END,
             updated_at = $10
         WHERE org_id = $1 AND agent_id = $2
         RETURNING *`,
        [
          input.orgId,
          input.agentId,
          input.profileName !== undefined,
          input.profileName === undefined ? null : normalizeText(input.profileName, CHANNEL_TEXT_MAX_LENGTH, 'Headless agent profile'),
          input.name !== undefined,
          input.name === undefined ? null : normalizeText(input.name, CHANNEL_TEXT_MAX_LENGTH, 'Headless agent name'),
          input.status || null,
          input.managed !== undefined,
          input.managed ?? null,
          nowIso(input.updatedAt),
        ],
      )
      if (!result.rows[0]) return null
      const agent = headlessAgentFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: input.orgId,
        accountId: input.actor?.accountId || null,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: 'headless_agent.updated',
        targetType: 'headless_agent',
        targetId: agent.agentId,
        metadata: {
          profileName: agent.profileName,
          name: agent.name,
          status: agent.status,
          managed: agent.managed,
        },
        createdAt: input.updatedAt,
      })
      return agent
    })
  }

  async getHeadlessAgent(orgId: string, agentId: string) {
    const row = await this.maybeOne(`SELECT * FROM headless_agents WHERE org_id = $1 AND agent_id = $2`, [orgId, agentId])
    return row ? headlessAgentFromRow(row) : null
  }

  async listHeadlessAgents(orgId: string) {
    const result = await this.pool.query(
      `SELECT * FROM headless_agents WHERE org_id = $1 ORDER BY updated_at DESC, agent_id`,
      [orgId],
    )
    return result.rows.map(headlessAgentFromRow)
  }

  async createChannelBinding(input: CreateChannelBindingInput) {
    return this.withTransaction(async (client) => {
      const now = nowIso(input.createdAt)
      const bindingLimit = input.quota?.maxGatewayChannelBindingsPerOrg
      if (bindingLimit && bindingLimit > 0) {
        await this.lockQuota(client, input.orgId, 'gateway_channel_bindings')
        const countRow = await this.one(
          `SELECT count(*)::int AS count
           FROM cloud_channel_bindings
           WHERE org_id = $1 AND status <> 'disabled'`,
          [input.orgId],
          client,
        )
        const activeBindings = numberValue(countRow.count)
        if (activeBindings >= bindingLimit) {
          throw new ControlPlaneQuotaExceededError({
            message: 'Gateway channel binding quota exceeded.',
            policyCode: input.quota?.policyCode || 'quota.gateway_channel_bindings_exceeded',
            retryAfterMs: 60_000,
            limit: bindingLimit,
            used: activeBindings,
            resetAt: new Date(Date.now() + 60_000).toISOString(),
          })
        }
      }
      const result = await client.query(
        `INSERT INTO cloud_channel_bindings (
          binding_id, org_id, agent_id, provider, external_workspace_id,
          display_name, status, credential_ref, settings, created_at, updated_at
         )
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10, $10)
         ON CONFLICT (binding_id) DO NOTHING
         RETURNING *`,
        [
          normalizeText(input.bindingId, CHANNEL_TEXT_MAX_LENGTH, 'Channel binding id'),
          input.orgId,
          input.agentId,
          normalizeProvider(input.provider),
          normalizeNullableText(input.externalWorkspaceId, CHANNEL_TEXT_MAX_LENGTH, 'External workspace id'),
          normalizeText(input.displayName, CHANNEL_TEXT_MAX_LENGTH, 'Channel binding name'),
          input.status || 'active',
          normalizeNullableText(input.credentialRef, CHANNEL_TEXT_MAX_LENGTH, 'Credential ref'),
          JSON.stringify(normalizeRecord(input.settings, 'Channel binding settings')),
          now,
        ],
      )
      const row = result.rows[0] || await this.one(
        `SELECT * FROM cloud_channel_bindings WHERE org_id = $1 AND binding_id = $2`,
        [input.orgId, input.bindingId],
        client,
      )
      const binding = channelBindingFromRow(row)
      if (result.rows[0]) {
        await this.recordAuditEventWithExecutor(client, {
          orgId: binding.orgId,
          actorType: 'system',
          actorId: 'channel_binding.create',
          eventType: 'channel_binding.created',
          targetType: 'channel_binding',
          targetId: binding.bindingId,
          metadata: { provider: binding.provider, displayName: binding.displayName, credentialRefConfigured: Boolean(binding.credentialRef) },
          createdAt: input.createdAt,
        })
      }
      return binding
    })
  }

  async updateChannelBinding(input: UpdateChannelBindingInput) {
    return this.withTransaction(async (client) => {
      const result = await client.query(
        `UPDATE cloud_channel_bindings
         SET display_name = CASE WHEN $3::boolean THEN $4 ELSE display_name END,
             status = COALESCE($5, status),
             credential_ref = CASE WHEN $6::boolean THEN $7 ELSE credential_ref END,
             settings = CASE WHEN $8::boolean THEN $9::jsonb ELSE settings END,
             updated_at = $10
         WHERE org_id = $1 AND binding_id = $2
         RETURNING *`,
        [
          input.orgId,
          input.bindingId,
          input.displayName !== undefined,
          input.displayName === undefined ? null : normalizeText(input.displayName, CHANNEL_TEXT_MAX_LENGTH, 'Channel binding name'),
          input.status || null,
          input.credentialRef !== undefined,
          input.credentialRef === undefined ? null : normalizeNullableText(input.credentialRef, CHANNEL_TEXT_MAX_LENGTH, 'Credential ref'),
          input.settings !== undefined,
          input.settings === undefined ? null : JSON.stringify(normalizeRecord(input.settings, 'Channel binding settings')),
          nowIso(input.updatedAt),
        ],
      )
      if (!result.rows[0]) return null
      const binding = channelBindingFromRow(result.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: input.orgId,
        accountId: input.actor?.accountId || null,
        actorType: input.actor?.actorType || 'system',
        actorId: input.actor?.actorId || null,
        eventType: 'channel_binding.updated',
        targetType: 'channel_binding',
        targetId: binding.bindingId,
        metadata: {
          provider: binding.provider,
          displayName: binding.displayName,
          status: binding.status,
          credentialRefConfigured: Boolean(binding.credentialRef),
          settingsChanged: input.settings !== undefined,
        },
        createdAt: input.updatedAt,
      })
      return binding
    })
  }

  async getChannelBinding(orgId: string, bindingId: string) {
    const row = await this.maybeOne(`SELECT * FROM cloud_channel_bindings WHERE org_id = $1 AND binding_id = $2`, [orgId, bindingId])
    return row ? channelBindingFromRow(row) : null
  }

  async listChannelBindings(orgId: string, agentId?: string | null) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_channel_bindings
       WHERE org_id = $1 AND ($2::text IS NULL OR agent_id = $2)
       ORDER BY updated_at DESC, binding_id`,
      [orgId, agentId || null],
    )
    return result.rows.map(channelBindingFromRow)
  }

  async upsertChannelIdentity(input: UpsertChannelIdentityInput) {
    const now = nowIso(input.updatedAt)
    const provider = normalizeProvider(input.provider)
    const externalWorkspaceId = normalizeNullableText(input.externalWorkspaceId, CHANNEL_TEXT_MAX_LENGTH, 'External workspace id')
    const externalUserId = normalizeText(input.externalUserId, CHANNEL_TEXT_MAX_LENGTH, 'External user id')
    const result = await this.pool.query(
      `INSERT INTO cloud_channel_identities (
        identity_id, org_id, provider, external_workspace_id, external_user_id,
        account_id, role, status, metadata, created_at, updated_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10, $10)
       ON CONFLICT (org_id, provider, (COALESCE(external_workspace_id, '')), external_user_id) DO UPDATE
       SET account_id = CASE
             WHEN EXCLUDED.account_id IS NOT NULL THEN EXCLUDED.account_id
             ELSE cloud_channel_identities.account_id
           END,
           role = EXCLUDED.role,
           status = EXCLUDED.status,
           metadata = EXCLUDED.metadata,
           updated_at = EXCLUDED.updated_at
       RETURNING *`,
      [
        input.identityId || `chid_${provider}_${externalUserId}`,
        input.orgId,
        provider,
        externalWorkspaceId,
        externalUserId,
        input.accountId || null,
        input.role || 'viewer',
        input.status || 'pending',
        JSON.stringify(normalizeRecord(input.metadata, 'Channel identity metadata')),
        now,
      ],
    )
    return channelIdentityFromRow(result.rows[0])
  }

  async getChannelIdentity(orgId: string, identityId: string) {
    const row = await this.maybeOne(`SELECT * FROM cloud_channel_identities WHERE org_id = $1 AND identity_id = $2`, [orgId, identityId])
    return row ? channelIdentityFromRow(row) : null
  }

  async findChannelIdentity(input: { orgId: string, provider: ChannelProviderId, externalWorkspaceId?: string | null, externalUserId: string }) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_channel_identities
       WHERE org_id = $1
         AND provider = $2
         AND COALESCE(external_workspace_id, '') = COALESCE($3, '')
         AND external_user_id = $4`,
      [
        input.orgId,
        normalizeProvider(input.provider),
        normalizeNullableText(input.externalWorkspaceId, CHANNEL_TEXT_MAX_LENGTH, 'External workspace id'),
        normalizeText(input.externalUserId, CHANNEL_TEXT_MAX_LENGTH, 'External user id'),
      ],
    )
    return row ? channelIdentityFromRow(row) : null
  }

  async bindChannelSession(input: BindChannelSessionInput) {
    const provider = normalizeProvider(input.provider)
    const externalWorkspaceId = normalizeNullableText(input.externalWorkspaceId, CHANNEL_TEXT_MAX_LENGTH, 'External workspace id')
    const relationship = await this.maybeOne(
      `SELECT b.binding_id
       FROM cloud_channel_bindings b
       JOIN headless_agents a
         ON a.agent_id = $3
        AND a.org_id = $1
       JOIN cloud_orgs o
         ON o.org_id = $1
       JOIN cloud_sessions s
         ON s.tenant_id = o.tenant_id
        AND s.session_id = $4
       WHERE b.org_id = $1
         AND b.binding_id = $2
         AND b.agent_id = a.agent_id
         AND b.provider = $5`,
      [input.orgId, input.channelBindingId, input.agentId, input.sessionId, provider],
    )
    if (!relationship) throw new Error('Channel session binding references must belong to the same org, agent, provider, and session.')
    const now = nowIso(input.createdAt)
    const result = await this.pool.query(
      `INSERT INTO cloud_channel_session_bindings (
        binding_id, org_id, agent_id, channel_binding_id, provider,
        external_workspace_id, external_thread_id, external_chat_id, session_id,
        last_event_sequence, last_workspace_sequence, last_chat_message_id,
        status, created_at, updated_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $14)
       ON CONFLICT DO NOTHING
       RETURNING *`,
      [
        normalizeText(input.bindingId, CHANNEL_TEXT_MAX_LENGTH, 'Channel session binding id'),
        input.orgId,
        input.agentId,
        input.channelBindingId,
        provider,
        externalWorkspaceId,
        normalizeText(input.externalThreadId, CHANNEL_TEXT_MAX_LENGTH, 'External thread id'),
        normalizeText(input.externalChatId, CHANNEL_TEXT_MAX_LENGTH, 'External chat id'),
        input.sessionId,
        normalizeNonNegativeInteger(input.lastEventSequence, 'Last event sequence'),
        normalizeNonNegativeInteger(input.lastWorkspaceSequence, 'Last workspace sequence'),
        normalizeNullableText(input.lastChatMessageId, CHANNEL_TEXT_MAX_LENGTH, 'Last chat message id'),
        input.status || 'active',
        now,
      ],
    )
    const row = result.rows[0] || await this.one(
      `SELECT * FROM cloud_channel_session_bindings
       WHERE org_id = $1 AND provider = $2 AND COALESCE(external_workspace_id, '') = COALESCE($3::text, '')
         AND external_chat_id = $4 AND external_thread_id = $5`,
      [input.orgId, provider, externalWorkspaceId, input.externalChatId, input.externalThreadId],
    )
    const binding = channelSessionBindingFromRow(row)
    if (result.rows[0]) {
      await this.recordAuditEvent({
        orgId: binding.orgId,
        actorType: 'system',
        actorId: 'channel_session.bind',
        eventType: 'channel_session_bound',
        targetType: 'channel_session_binding',
        targetId: binding.bindingId,
        metadata: { provider: binding.provider, sessionId: binding.sessionId },
        createdAt: input.createdAt,
      })
    }
    return binding
  }

  async getChannelSessionBinding(orgId: string, bindingId: string) {
    const row = await this.maybeOne(`SELECT * FROM cloud_channel_session_bindings WHERE org_id = $1 AND binding_id = $2`, [orgId, bindingId])
    return row ? channelSessionBindingFromRow(row) : null
  }

  async findChannelSessionBindingByThread(input: { orgId: string, provider: ChannelProviderId, externalWorkspaceId?: string | null, externalChatId: string, externalThreadId: string }) {
    const externalWorkspaceId = normalizeNullableText(input.externalWorkspaceId, CHANNEL_TEXT_MAX_LENGTH, 'External workspace id')
    const row = await this.maybeOne(
      `SELECT * FROM cloud_channel_session_bindings
       WHERE org_id = $1 AND provider = $2 AND COALESCE(external_workspace_id, '') = COALESCE($3::text, '')
         AND external_chat_id = $4 AND external_thread_id = $5`,
      [input.orgId, normalizeProvider(input.provider), externalWorkspaceId, input.externalChatId, input.externalThreadId],
    )
    return row ? channelSessionBindingFromRow(row) : null
  }

  async listChannelSessionBindingsForSession(orgId: string, sessionId: string) {
    const result = await this.pool.query(
      `SELECT * FROM cloud_channel_session_bindings
       WHERE org_id = $1 AND session_id = $2 AND status = 'active'
       ORDER BY updated_at DESC, binding_id`,
      [orgId, sessionId],
    )
    return result.rows.map(channelSessionBindingFromRow)
  }

  async updateChannelCursor(input: UpdateChannelCursorInput) {
    const lastEventSequence = normalizeNonNegativeInteger(input.lastEventSequence, 'Last event sequence'), lastWorkspaceSequence = normalizeNonNegativeInteger(input.lastWorkspaceSequence, 'Last workspace sequence')
    const result = await this.pool.query(
      `UPDATE cloud_channel_session_bindings
       SET last_event_sequence = $3,
           last_workspace_sequence = $4,
           last_chat_message_id = CASE WHEN $5::boolean THEN $6 ELSE last_chat_message_id END,
           updated_at = $7
       WHERE org_id = $1
         AND binding_id = $2
         AND last_event_sequence <= $3
         AND last_workspace_sequence <= $4
       RETURNING *`,
      [
        input.orgId,
        input.bindingId,
        lastEventSequence,
        lastWorkspaceSequence,
        input.lastChatMessageId !== undefined,
        normalizeNullableText(input.lastChatMessageId, CHANNEL_TEXT_MAX_LENGTH, 'Last chat message id'),
        nowIso(input.updatedAt),
      ],
    )
    if (result.rows[0]) return { ok: true as const, binding: channelSessionBindingFromRow(result.rows[0]) }
    const existing = await this.maybeOne(`SELECT * FROM cloud_channel_session_bindings WHERE org_id = $1 AND binding_id = $2`, [input.orgId, input.bindingId])
    if (!existing) return { ok: false as const, reason: 'not_found' as const }
    return { ok: false as const, reason: 'stale' as const, binding: channelSessionBindingFromRow(existing) }
  }

  async createChannelInteraction(input: CreateChannelInteractionInput): Promise<IssuedChannelInteractionRecord> {
    const plaintextToken = generateChannelInteractionToken({ interactionId: input.interactionId, secret: input.tokenSecret })
    const relationship = await this.maybeOne(
      `SELECT a.agent_id
       FROM headless_agents a
       JOIN cloud_orgs o
         ON o.org_id = $1
       JOIN cloud_sessions s
         ON s.tenant_id = o.tenant_id
        AND s.session_id = $3
       LEFT JOIN cloud_channel_identities i
         ON i.identity_id = $4
       WHERE a.org_id = $1
         AND a.agent_id = $2
         AND ($4::text IS NULL OR i.org_id = $1)`,
      [input.orgId, input.agentId, input.sessionId, input.createdByIdentityId || null],
    )
    if (!relationship) throw new Error('Channel interaction references must belong to the same org, agent, session, and identity.')
    const result = await this.pool.query(
      `INSERT INTO cloud_channel_interactions (
        interaction_id, org_id, agent_id, session_id, provider, external_interaction_id,
        token_hash, kind, target_id, status, created_by_identity_id,
        expires_at, used_at, created_at, updated_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending', $10, $11, NULL, $12, $12)
       ON CONFLICT (interaction_id) DO NOTHING
       RETURNING *`,
      [
        input.interactionId,
        input.orgId,
        input.agentId,
        input.sessionId,
        normalizeProvider(input.provider),
        normalizeNullableText(input.externalInteractionId, CHANNEL_TEXT_MAX_LENGTH, 'External interaction id'),
        hashChannelInteractionToken(plaintextToken),
        input.kind,
        normalizeText(input.targetId, CHANNEL_TEXT_MAX_LENGTH, 'Interaction target id'),
        input.createdByIdentityId || null,
        input.expiresAt.toISOString(),
        nowIso(input.createdAt),
      ],
    )
    if (!result.rows[0]) throw new Error(`Channel interaction ${input.interactionId} already exists.`)
    return { interaction: channelInteractionFromRow(result.rows[0]), plaintextToken }
  }

  async findChannelInteraction(input: FindChannelInteractionInput) {
    const tokenHash = input.token ? hashChannelInteractionToken(input.token) : null
    const now = nowIso(input.now)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_channel_interactions
       WHERE org_id = $1
         AND status = 'pending'
         AND expires_at > $5
         AND (
           ($2::text IS NOT NULL AND token_hash = $2)
           OR (
             $3::text IS NOT NULL
             AND $4::text IS NOT NULL
             AND provider = $3
             AND external_interaction_id = $4
           )
         )`,
      [input.orgId, tokenHash, input.provider || null, input.externalInteractionId || null, now],
    )
    return row ? channelInteractionFromRow(row) : null
  }

  async resolveChannelInteraction(input: ResolveChannelInteractionInput) {
    const tokenHash = input.token ? hashChannelInteractionToken(input.token) : null
    const now = nowIso(input.usedAt)
    const result = await this.pool.query(
      `UPDATE cloud_channel_interactions
       SET status = 'used', used_at = $5, updated_at = $5
       WHERE org_id = $1
         AND status = 'pending'
         AND expires_at > $5
         AND (
           ($2::text IS NOT NULL AND token_hash = $2)
           OR (
             $3::text IS NOT NULL
             AND $4::text IS NOT NULL
             AND provider = $3
             AND external_interaction_id = $4
           )
         )
       RETURNING *`,
      [input.orgId, tokenHash, input.provider || null, input.externalInteractionId || null, now],
    )
    const interaction = result.rows[0] ? channelInteractionFromRow(result.rows[0]) : null
    if (interaction) {
      await this.recordAuditEvent({
        orgId: interaction.orgId,
        actorType: 'system',
        actorId: input.identityId,
        eventType: 'channel_interaction.used',
        targetType: 'channel_interaction',
        targetId: interaction.interactionId,
        metadata: { kind: interaction.kind, targetId: interaction.targetId },
        createdAt: input.usedAt,
      })
    }
    return interaction
  }

  async resolveChannelInteractionWithCommand(input: ResolveChannelInteractionWithCommandInput) {
    return this.withTransaction(async (client) => {
      const tokenHash = input.token ? hashChannelInteractionToken(input.token) : null
      const usedAt = nowIso(input.usedAt)
      const selected = await this.maybeOne(
        `SELECT * FROM cloud_channel_interactions
         WHERE org_id = $1
           AND status = 'pending'
           AND expires_at > $5
           AND (
             ($2::text IS NOT NULL AND token_hash = $2)
             OR (
               $3::text IS NOT NULL
               AND $4::text IS NOT NULL
               AND provider = $3
               AND external_interaction_id = $4
             )
           )
         FOR UPDATE`,
        [input.orgId, tokenHash, input.provider || null, input.externalInteractionId || null, usedAt],
        client,
      )
      if (!selected) return null
      const interaction = channelInteractionFromRow(selected)
      if (input.command.sessionId !== interaction.sessionId) {
        throw new Error('Channel interaction command session does not match interaction session.')
      }
      await this.requireTenantUser(input.command.tenantId, input.command.userId, client)
      await this.requireSession(input.command.tenantId, input.command.sessionId, client, true)
      const payload = input.command.payload || {}
      const existingCommand = await this.maybeOne(
        `SELECT * FROM cloud_session_commands WHERE command_id = $1`,
        [input.command.commandId],
        client,
      )
      let command: SessionCommandRecord
      if (existingCommand) {
        command = commandFromRow(existingCommand)
        if (
          command.tenantId !== input.command.tenantId
          || command.userId !== input.command.userId
          || command.sessionId !== input.command.sessionId
          || command.kind !== input.command.kind
          || command.targetLeaseToken !== (input.command.targetLeaseToken ?? null)
          || stableJson(command.payload) !== stableJson(payload)
        ) {
          throw new Error(`Command id ${input.command.commandId} was reused with different content.`)
        }
      } else {
        const createdAt = nowIso(input.command.createdAt)
        const sequence = await this.incrementSessionCounter(
          client,
          input.command.tenantId,
          input.command.sessionId,
          'next_command_sequence',
        )
        const commandResult = await client.query(
          `INSERT INTO cloud_session_commands (
            command_id, tenant_id, user_id, session_id, kind, payload,
            target_lease_token, created_sequence, created_at, status
           )
           VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, $9, 'pending')
           RETURNING *`,
          [
            input.command.commandId,
            input.command.tenantId,
            input.command.userId,
            input.command.sessionId,
            input.command.kind,
            JSON.stringify(payload),
            input.command.targetLeaseToken ?? null,
            sequence,
            createdAt,
          ],
        )
        command = commandFromRow(commandResult.rows[0])
      }
      const updated = await client.query(
        `UPDATE cloud_channel_interactions
         SET status = 'used', used_at = $2, updated_at = $2
         WHERE interaction_id = $1
         RETURNING *`,
        [interaction.interactionId, usedAt],
      )
      const resolvedInteraction = channelInteractionFromRow(updated.rows[0])
      await this.recordAuditEventWithExecutor(client, {
        orgId: resolvedInteraction.orgId,
        actorType: 'system',
        actorId: input.identityId,
        eventType: 'channel_interaction.used',
        targetType: 'channel_interaction',
        targetId: resolvedInteraction.interactionId,
        metadata: { kind: resolvedInteraction.kind, targetId: resolvedInteraction.targetId },
        createdAt: input.usedAt,
      })
      return { interaction: resolvedInteraction, command }
    })
  }

  async createChannelDelivery(input: CreateChannelDeliveryInput) {
    return this.channelDeliveries.create(input)
  }

  async listChannelDeliveries(input: ListChannelDeliveriesInput) {
    return this.channelDeliveries.list(input)
  }

  async claimNextChannelDelivery(input: ClaimChannelDeliveryInput) {
    return this.channelDeliveries.claimNext(input)
  }

  async ackChannelDelivery(input: AckChannelDeliveryInput) {
    return this.channelDeliveries.ack(input)
  }

  async createCloudCoordinationWatch(input: CreateCloudCoordinationWatchInput): Promise<CoordinationWatch> {
    return this.coordinationWatches.create(input)
  }

  async updateCloudCoordinationWatch(input: UpdateCloudCoordinationWatchInput): Promise<CoordinationWatch | null> {
    return this.coordinationWatches.update(input)
  }

  async getCloudCoordinationWatch(workspaceId: string, watchId: string): Promise<CoordinationWatch | null> {
    return this.coordinationWatches.get(workspaceId, watchId)
  }

  async listCloudCoordinationWatches(input: ListCloudCoordinationWatchesInput): Promise<CoordinationWatch[]> {
    return this.coordinationWatches.list(input)
  }

  async listMatchingCloudCoordinationWatches(input: ListMatchingCloudCoordinationWatchesInput): Promise<CoordinationWatch[]> {
    return this.coordinationWatches.listMatching(input)
  }

  async deleteCloudCoordinationWatch(workspaceId: string, watchId: string): Promise<boolean> {
    return this.coordinationWatches.delete(workspaceId, watchId)
  }

  async claimChannelProviderEvent(input: ClaimChannelProviderEventInput) {
    return this.channelProviderEvents.claim(input)
  }

  async completeChannelProviderEvent(input: CompleteChannelProviderEventInput) {
    return this.channelProviderEvents.complete(input)
  }

  async createSession(input: {
    tenantId: string
    userId: string
    sessionId: string
    opencodeSessionId: string
    profileName: string
    title?: string | null
    createdAt?: Date
    quota?: {
      orgId?: string | null
      maxConcurrentSessionsPerOrg?: number | null
      policyCode?: string
    } | null
  }) {
    return this.withTransaction(async (client) => {
      await this.requireTenantUser(input.tenantId, input.userId, client)
      const existing = await this.maybeOne(
        `SELECT * FROM cloud_sessions WHERE tenant_id = $1 AND session_id = $2`,
        [input.tenantId, input.sessionId],
        client,
      )
      if (existing) return sessionFromRow(existing)
      await assertPostgresConcurrentSessionQuota(client, { tenantId: input.tenantId, quota: input.quota, now: input.createdAt }, this.quotaDeps)
      const createdAt = nowIso(input.createdAt)
      const result = await client.query(
        `INSERT INTO cloud_sessions (
          tenant_id, session_id, user_id, opencode_session_id, profile_name,
          status, title, created_at, updated_at
         )
         VALUES ($1, $2, $3, $4, $5, 'idle', $6, $7, $7)
         RETURNING *`,
        [
          input.tenantId,
          input.sessionId,
          input.userId,
          input.opencodeSessionId,
          input.profileName,
          input.title || null,
          createdAt,
        ],
      )
      return sessionFromRow(result.rows[0])
    })
  }

  async getSession(tenantId: string, userId: string, sessionId: string) {
    await this.requireTenantUser(tenantId, userId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_sessions
       WHERE tenant_id = $1 AND user_id = $2 AND session_id = $3`,
      [tenantId, userId, sessionId],
    )
    return row ? sessionFromRow(row) : null
  }

  async getSessionForTenant(tenantId: string, sessionId: string) {
    await this.requireTenant(tenantId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_sessions WHERE tenant_id = $1 AND session_id = $2`,
      [tenantId, sessionId],
    )
    return row ? sessionFromRow(row) : null
  }

  async findSession(sessionId: string) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_sessions
       WHERE session_id = $1 OR opencode_session_id = $1
       ORDER BY updated_at DESC
       LIMIT 1`,
      [sessionId],
    )
    return row ? sessionFromRow(row) : null
  }

  async listSessions(tenantId: string, userId: string) {
    await this.requireTenantUser(tenantId, userId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_sessions
       WHERE tenant_id = $1 AND user_id = $2
       ORDER BY updated_at DESC, session_id`,
      [tenantId, userId],
    )
    return result.rows.map(sessionFromRow)
  }

  async listSessionsPage(input: ListSessionsPageInput) {
    await this.requireTenantUser(input.tenantId, input.userId)
    const limit = Math.max(1, Math.min(500, Math.floor(input.limit ?? 100)))
    const cursor = decodeSessionPageCursor(input.cursor, input)
    const params: unknown[] = [input.tenantId, input.userId]
    const where = ['tenant_id = $1', 'user_id = $2']
    if (input.status) {
      params.push(input.status)
      where.push(`status = $${params.length}`)
    }
    if (input.profileName) {
      params.push(input.profileName)
      where.push(`profile_name = $${params.length}`)
    }
    const query = input.query?.trim().toLowerCase()
    if (query) {
      params.push(`%${query}%`)
      where.push(`(
        lower(COALESCE(title, '')) LIKE $${params.length}
        OR lower(session_id) LIKE $${params.length}
        OR lower(opencode_session_id) LIKE $${params.length}
        OR lower(profile_name) LIKE $${params.length}
      )`)
    }
    if (cursor) {
      params.push(cursor.updatedAt, cursor.sessionId)
      const updatedAtParam = params.length - 1
      const sessionIdParam = params.length
      where.push(`(updated_at < $${updatedAtParam} OR (updated_at = $${updatedAtParam} AND session_id > $${sessionIdParam}))`)
    }
    params.push(limit + 1)
    const result = await this.pool.query(
      `SELECT * FROM cloud_sessions
       WHERE ${where.join(' AND ')}
       ORDER BY updated_at DESC, session_id
       LIMIT $${params.length}`,
      params,
    )
    const rows = result.rows.map(sessionFromRow)
    const items = rows.slice(0, limit)
    return {
      items,
      nextCursor: rows.length > limit && items.length > 0 ? encodeSessionPageCursor(items[items.length - 1]!, input) : null,
      totalEstimate: rows.length > limit ? limit + 1 : rows.length,
    }
  }

  async listAllSessions() {
    const result = await this.pool.query(
      `SELECT * FROM cloud_sessions ORDER BY updated_at DESC, tenant_id, session_id`,
    )
    return result.rows.map(sessionFromRow)
  }

  async listRunnableSessions(input: {
    limit?: number | null
    now?: Date
  } = {}) {
    return listPostgresRunnableSessions(this.pool, input)
  }

  async claimRunnableSessions(input: {
    workerId: string
    limit?: number | null
    now?: Date
    ttlMs?: number
  }) {
    const now = input.now || new Date()
    const nowMs = now.getTime()
    const ttlMs = input.ttlMs ?? 30_000
    const limit = Math.max(1, Math.min(1_000, Math.floor(input.limit ?? 100)))
    return this.withTransaction(async (client) => {
      const selected = await client.query(
        `SELECT sessions.*, runnable.first_sequence
         FROM cloud_sessions sessions
         JOIN (
           SELECT commands.tenant_id, commands.session_id, min(commands.created_sequence) AS first_sequence
           FROM cloud_session_commands commands
           LEFT JOIN cloud_worker_leases leases
             ON leases.tenant_id = commands.tenant_id
            AND leases.session_id = commands.session_id
           WHERE commands.target_lease_token IS NULL
             AND commands.status IN ('pending', 'running')
             AND (commands.status <> 'pending' OR commands.available_at IS NULL OR commands.available_at <= $2)
             AND (leases.lease_expires_at_ms IS NULL OR leases.lease_expires_at_ms <= $1)
           GROUP BY commands.tenant_id, commands.session_id
           ORDER BY first_sequence, commands.tenant_id, commands.session_id
           LIMIT $3
         ) runnable
           ON runnable.tenant_id = sessions.tenant_id
          AND runnable.session_id = sessions.session_id
         ORDER BY runnable.first_sequence, sessions.tenant_id, sessions.session_id
         FOR UPDATE OF sessions SKIP LOCKED`,
        [nowMs, now.toISOString(), limit],
      )
      const leases = []
      for (const row of selected.rows) {
        const tenantId = String(row.tenant_id)
        const sessionId = String(row.session_id)
        const currentLease = await this.getLease(tenantId, sessionId, client, true)
        if (currentLease && currentLease.leaseExpiresAt > nowMs) continue
        const attempt = numberValue(row.next_lease_attempt) + 1
        const leaseRecord = {
          tenantId,
          sessionId,
          leasedBy: input.workerId,
          leaseToken: `${tenantId}:${sessionId}:${attempt}:${input.workerId}`,
          leaseExpiresAt: nowMs + ttlMs,
          checkpointVersion: currentLease?.checkpointVersion || 0,
        }
        await client.query(
          `UPDATE cloud_sessions
           SET next_lease_attempt = $3, status = 'running', updated_at = $4
           WHERE tenant_id = $1 AND session_id = $2`,
          [tenantId, sessionId, attempt, now.toISOString()],
        )
        const result = await client.query(
          `INSERT INTO cloud_worker_leases (
            tenant_id, session_id, leased_by, lease_token, lease_expires_at_ms, checkpoint_version
           )
           VALUES ($1, $2, $3, $4, $5, $6)
           ON CONFLICT (tenant_id, session_id) DO UPDATE
           SET leased_by = EXCLUDED.leased_by,
               lease_token = EXCLUDED.lease_token,
               lease_expires_at_ms = EXCLUDED.lease_expires_at_ms,
               checkpoint_version = EXCLUDED.checkpoint_version
           RETURNING *`,
          [
            tenantId,
            sessionId,
            leaseRecord.leasedBy,
            leaseRecord.leaseToken,
            leaseRecord.leaseExpiresAt,
            leaseRecord.checkpointVersion,
          ],
        )
        leases.push(leaseFromRow(result.rows[0]))
      }
      return {
        leases,
        pendingSessionCountEstimate: selected.rows.length,
      }
    })
  }

  async bindSessionRuntime(input: {
    tenantId: string
    sessionId: string
    opencodeSessionId: string
    title?: string | null
    leaseToken?: string | null
    updatedAt?: Date
  }) {
    return this.withTransaction(async (client) => {
      await this.assertLeaseTokenIfPresent(input.tenantId, input.sessionId, input.leaseToken, client)
      const updatedAt = nowIso(input.updatedAt)
      const result = await client.query(
        `UPDATE cloud_sessions
         SET opencode_session_id = $3,
             title = CASE WHEN $4::boolean THEN $5 ELSE title END,
             updated_at = $6
         WHERE tenant_id = $1 AND session_id = $2
         RETURNING *`,
        [
          input.tenantId,
          input.sessionId,
          input.opencodeSessionId,
          input.title !== undefined,
          input.title ?? null,
          updatedAt,
        ],
      )
      if (!result.rows[0]) throw new Error(`Unknown session ${input.sessionId}.`)
      return sessionFromRow(result.rows[0])
    })
  }

  async updateSessionStatus(input: {
    tenantId: string
    sessionId: string
    status: ControlPlaneSessionStatus
    title?: string | null
    leaseToken?: string | null
    updatedAt?: Date
  }) {
    return this.withTransaction(async (client) => {
      await this.assertLeaseTokenIfPresent(input.tenantId, input.sessionId, input.leaseToken, client)
      const updatedAt = nowIso(input.updatedAt)
      const result = await client.query(
        `UPDATE cloud_sessions
         SET status = $3,
             title = CASE WHEN $4::boolean THEN $5 ELSE title END,
             updated_at = $6
         WHERE tenant_id = $1 AND session_id = $2
         RETURNING *`,
        [
          input.tenantId,
          input.sessionId,
          input.status,
          input.title !== undefined,
          input.title ?? null,
          updatedAt,
        ],
      )
      if (!result.rows[0]) throw new Error(`Unknown session ${input.sessionId}.`)
      return sessionFromRow(result.rows[0])
    })
  }

  async appendSessionEvent(input: {
    tenantId: string
    sessionId: string
    eventId?: string
    type: string
    payload?: Record<string, unknown>
    leaseToken?: string | null
    createdAt?: Date
  }) {
    return this.withTransaction(async (client) => {
      await this.requireSession(input.tenantId, input.sessionId, client, true)
      await this.assertLeaseTokenIfPresent(input.tenantId, input.sessionId, input.leaseToken, client)
      const payload = input.payload || {}
      if (input.eventId) {
        const existing = await this.findEvent(input.tenantId, input.sessionId, input.eventId, client)
        if (existing) return this.replayOrRejectEvent(existing, input.type, payload)
      }
      const createdAt = nowIso(input.createdAt)
      const sequence = await this.incrementSessionCounter(
        client,
        input.tenantId,
        input.sessionId,
        'next_event_sequence',
        createdAt,
      )
      const eventId = input.eventId || `${input.sessionId}:${sequence}`
      const inserted = await client.query(
        `INSERT INTO cloud_session_events (
          tenant_id, session_id, event_id, sequence, type, payload, created_at
         )
         VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7)
         RETURNING *`,
        [input.tenantId, input.sessionId, eventId, sequence, input.type, JSON.stringify(payload), createdAt],
      )
      return eventFromRow(inserted.rows[0])
    })
  }

  async listSessionEvents(tenantId: string, sessionId: string, afterSequence = 0) {
    await this.requireSession(tenantId, sessionId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_session_events
       WHERE tenant_id = $1 AND session_id = $2 AND sequence > $3
       ORDER BY sequence`,
      [tenantId, sessionId, afterSequence],
    )
    return result.rows.map(eventFromRow)
  }

  async appendWorkspaceEvent(input: AppendWorkspaceEventInput) {
    return this.withTransaction(async (client) => {
      await this.requireTenantUser(input.tenantId, input.userId, client)
      if (input.sessionId) {
        const session = await this.requireSession(input.tenantId, input.sessionId, client, true)
        if (String(session.user_id) !== input.userId) {
          throw new Error(`Session ${input.sessionId} does not belong to user ${input.userId}.`)
        }
      }

      const payload = input.payload || {}
      const sessionId = input.sessionId || null
      const entityType = optionalTrimmedText(input.entityType) || (sessionId ? 'session' : 'workspace')
      const entityId = optionalTrimmedText(input.entityId) || sessionId || input.userId
      const operation = optionalTrimmedText(input.operation) || workspaceOperationFromType(input.type)
      await client.query(
        `INSERT INTO cloud_workspace_event_counters (tenant_id, user_id, next_sequence)
         VALUES ($1, $2, 0)
         ON CONFLICT (tenant_id, user_id) DO NOTHING`,
        [input.tenantId, input.userId],
      )
      const counter = await this.one(
        `SELECT next_sequence
         FROM cloud_workspace_event_counters
         WHERE tenant_id = $1 AND user_id = $2
         FOR UPDATE`,
        [input.tenantId, input.userId],
        client,
      )
      if (input.eventId) {
        const existing = await this.findWorkspaceEvent(input.tenantId, input.userId, input.eventId, client)
        if (existing) {
          const projectionVersion = Number.isFinite(input.projectionVersion)
            ? Math.max(0, Math.floor(input.projectionVersion || 0))
            : existing.projectionVersion
          return this.replayOrRejectWorkspaceEvent(existing, input.type, payload, {
            sessionId,
            entityType,
            entityId,
            operation,
            projectionVersion,
          })
        }
      }

      const sequence = numberValue(counter.next_sequence) + 1
      const eventId = input.eventId || `${input.userId}:${sequence}`
      const projectionVersion = Number.isFinite(input.projectionVersion)
        ? Math.max(0, Math.floor(input.projectionVersion || 0))
        : sequence
      await client.query(
        `UPDATE cloud_workspace_event_counters
         SET next_sequence = $3
         WHERE tenant_id = $1 AND user_id = $2`,
        [input.tenantId, input.userId, sequence],
      )
      const createdAt = nowIso(input.createdAt)
      const inserted = await client.query(
        `INSERT INTO cloud_workspace_events (
          tenant_id, user_id, event_id, sequence, session_id,
          entity_type, entity_id, operation, projection_version,
          type, payload, created_at
         )
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12)
         RETURNING *`,
        [
          input.tenantId,
          input.userId,
          eventId,
          sequence,
          sessionId,
          entityType,
          entityId,
          operation,
          projectionVersion,
          input.type,
          JSON.stringify(payload),
          createdAt,
        ],
      )
      return workspaceEventFromRow(inserted.rows[0])
    })
  }

  async listWorkspaceEvents(tenantId: string, userId: string, afterSequence = 0) {
    await this.requireTenantUser(tenantId, userId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_workspace_events
       WHERE tenant_id = $1 AND user_id = $2 AND sequence > $3
       ORDER BY sequence`,
      [tenantId, userId, afterSequence],
    )
    return result.rows.map(workspaceEventFromRow)
  }

  async getWorkspaceEventCursor(tenantId: string, userId: string): Promise<WorkspaceEventCursorRecord> {
    await this.requireTenantUser(tenantId, userId)
    const result = await this.pool.query(
      `SELECT min(sequence) AS earliest_sequence, max(sequence) AS latest_sequence
       FROM cloud_workspace_events
       WHERE tenant_id = $1 AND user_id = $2`,
      [tenantId, userId],
    )
    return workspaceEventCursorFromRow(result.rows[0])
  }

  async writeSessionProjection(input: {
    tenantId: string
    sessionId: string
    sequence: number
    view: Record<string, unknown>
    leaseToken?: string | null
    updatedAt?: Date
  }) {
    return this.withTransaction(async (client) => {
      await this.requireSession(input.tenantId, input.sessionId, client, true)
      await this.assertLeaseTokenIfPresent(input.tenantId, input.sessionId, input.leaseToken, client)
      const lease = await this.getLease(input.tenantId, input.sessionId, client, true)
      if (lease && lease.leaseToken !== (input.leaseToken ?? null)) {
        throw new Error('Projection write used a stale worker lease.')
      }
      const currentRow = await this.maybeOne(
        `SELECT * FROM cloud_session_projections WHERE tenant_id = $1 AND session_id = $2`,
        [input.tenantId, input.sessionId],
        client,
      )
      const current = currentRow ? projectionFromRow(currentRow) : null
      if (input.sequence < (current?.sequence || 0)) {
        throw new Error('Projection sequence must be monotonic.')
      }
      if (input.sequence === current?.sequence) {
        if (stableJson(current.view) !== stableJson(input.view)) {
          throw new Error('Projection sequence was reused with different content.')
        }
        return current
      }
      const updatedAt = nowIso(input.updatedAt)
      const result = await client.query(
        `INSERT INTO cloud_session_projections (tenant_id, session_id, sequence, view, updated_at)
         VALUES ($1, $2, $3, $4::jsonb, $5)
         ON CONFLICT (tenant_id, session_id) DO UPDATE
         SET sequence = EXCLUDED.sequence, view = EXCLUDED.view, updated_at = EXCLUDED.updated_at
         RETURNING *`,
        [input.tenantId, input.sessionId, input.sequence, JSON.stringify(input.view), updatedAt],
      )
      await client.query(
        `UPDATE cloud_sessions SET updated_at = $3 WHERE tenant_id = $1 AND session_id = $2`,
        [input.tenantId, input.sessionId, updatedAt],
      )
      return projectionFromRow(result.rows[0])
    })
  }

  async getSessionProjection(tenantId: string, sessionId: string) {
    await this.requireSession(tenantId, sessionId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_session_projections WHERE tenant_id = $1 AND session_id = $2`,
      [tenantId, sessionId],
    )
    return row ? projectionFromRow(row) : null
  }

  async claimSessionLease(
    tenantId: string,
    sessionId: string,
    workerId: string,
    now = new Date(),
    ttlMs = 30_000,
    quota: {
      orgId?: string | null
      maxActiveWorkersPerOrg?: number | null
      policyCode?: string
    } | null = null,
  ) {
    return this.withTransaction(async (client) => {
      const session = await this.requireSession(tenantId, sessionId, client, true)
      const lease = await this.getLease(tenantId, sessionId, client, true)
      const nowMs = now.getTime()
      if (lease && lease.leaseExpiresAt > nowMs) return null
      if (!(await checkPostgresActiveWorkerQuota(client, { tenantId, quota, nowMs }, this.quotaDeps))) return null
      const attempt = numberValue(session.next_lease_attempt) + 1
      const leaseRecord: WorkerLeaseRecord = {
        tenantId,
        sessionId,
        leasedBy: workerId,
        leaseToken: `${tenantId}:${sessionId}:${attempt}:${workerId}`,
        leaseExpiresAt: nowMs + ttlMs,
        checkpointVersion: lease?.checkpointVersion || 0,
      }
      await client.query(
        `UPDATE cloud_sessions
         SET next_lease_attempt = $3, status = 'running', updated_at = $4
         WHERE tenant_id = $1 AND session_id = $2`,
        [tenantId, sessionId, attempt, now.toISOString()],
      )
      const result = await client.query(
        `INSERT INTO cloud_worker_leases (
          tenant_id, session_id, leased_by, lease_token, lease_expires_at_ms, checkpoint_version
         )
         VALUES ($1, $2, $3, $4, $5, $6)
         ON CONFLICT (tenant_id, session_id) DO UPDATE
         SET leased_by = EXCLUDED.leased_by,
             lease_token = EXCLUDED.lease_token,
             lease_expires_at_ms = EXCLUDED.lease_expires_at_ms,
             checkpoint_version = EXCLUDED.checkpoint_version
         RETURNING *`,
        [
          tenantId,
          sessionId,
          leaseRecord.leasedBy,
          leaseRecord.leaseToken,
          leaseRecord.leaseExpiresAt,
          leaseRecord.checkpointVersion,
        ],
      )
      return leaseFromRow(result.rows[0])
    })
  }

  async releaseSessionLease(lease: WorkerLeaseRecord, now = new Date()) {
    return this.withTransaction(async (client) => {
      const result = await client.query(
        `DELETE FROM cloud_worker_leases
         WHERE tenant_id = $1
           AND session_id = $2
           AND lease_token = $3
         RETURNING lease_token`,
        [lease.tenantId, lease.sessionId, lease.leaseToken],
      )
      if (!result.rows[0]) return false
      await client.query(
        `UPDATE cloud_sessions
         SET status = 'idle', updated_at = $3
         WHERE tenant_id = $1
           AND session_id = $2`,
        [lease.tenantId, lease.sessionId, nowIso(now)],
      )
      return true
    })
  }

  async renewSessionLease(lease: WorkerLeaseRecord, now = new Date(), ttlMs = 30_000) {
    return this.withTransaction(async (client) => {
      await this.assertCurrentLease(lease, client)
      const result = await client.query(
        `UPDATE cloud_worker_leases
         SET lease_expires_at_ms = $3
         WHERE tenant_id = $1 AND session_id = $2
         RETURNING *`,
        [lease.tenantId, lease.sessionId, now.getTime() + ttlMs],
      )
      return leaseFromRow(result.rows[0])
    })
  }

  async checkpointSession(lease: WorkerLeaseRecord) {
    return this.withTransaction(async (client) => {
      const current = await this.assertCurrentLease(lease, client)
      if (lease.checkpointVersion !== current.checkpointVersion) {
        throw new Error('Checkpoint version is stale.')
      }
      const result = await client.query(
        `UPDATE cloud_worker_leases
         SET checkpoint_version = checkpoint_version + 1
         WHERE tenant_id = $1 AND session_id = $2
         RETURNING *`,
        [lease.tenantId, lease.sessionId],
      )
      return leaseFromRow(result.rows[0])
    })
  }

  async reapExpiredSessionLeases(input: ReapExpiredSessionLeasesInput = {}): Promise<ReapedSessionLeaseRecord[]> {
    const now = input.now || new Date()
    const nowMs = now.getTime()
    const nowIsoValue = now.toISOString()
    const maxAttempts = Math.max(1, Math.floor(input.maxCommandAttempts ?? 3))
    const limit = Math.max(1, Math.min(1_000, Math.floor(input.limit ?? 100)))
    return this.withTransaction(async (client) => {
      const expired = await client.query(
        `SELECT *
         FROM cloud_worker_leases
         WHERE lease_expires_at_ms <= $1
         ORDER BY lease_expires_at_ms ASC, tenant_id, session_id
         LIMIT $2
         FOR UPDATE SKIP LOCKED`,
        [nowMs, limit],
      )
      const reaped: ReapedSessionLeaseRecord[] = []
      for (const row of expired.rows) {
        const lease = leaseFromRow(row)
        const commands = await client.query(
          `SELECT *
           FROM cloud_session_commands
           WHERE tenant_id = $1
             AND session_id = $2
             AND status = 'running'
             AND claimed_lease_token = $3
           ORDER BY created_sequence
           FOR UPDATE`,
          [lease.tenantId, lease.sessionId, lease.leaseToken],
        )
        const retriedCommandIds: string[] = []
        const failedCommandIds: string[] = []
        for (const commandRow of commands.rows) {
          const command = commandFromRow(commandRow)
          if (command.attemptCount >= maxAttempts) {
            const summary = 'Worker lease expired after the maximum retry attempts.'
            await client.query(
              `UPDATE cloud_session_commands
               SET status = 'failed',
                   error = $2,
                   last_error_code = 'lease_expired_max_attempts',
                   last_error_summary = $2
               WHERE command_id = $1`,
              [command.commandId, summary],
            )
            failedCommandIds.push(command.commandId)
          } else {
            await client.query(
              `UPDATE cloud_session_commands
               SET status = 'pending',
                   claimed_by = NULL,
                   claimed_lease_token = NULL,
                   available_at = $2,
                   error = NULL,
                   last_error_code = 'lease_expired',
                   last_error_summary = 'Worker lease expired before command completion.'
               WHERE command_id = $1`,
              [command.commandId, nowIsoValue],
            )
            retriedCommandIds.push(command.commandId)
          }
        }
        await client.query(
          `DELETE FROM cloud_worker_leases
           WHERE tenant_id = $1 AND session_id = $2 AND lease_token = $3`,
          [lease.tenantId, lease.sessionId, lease.leaseToken],
        )
        const action: ReapedSessionLeaseRecord['action'] = failedCommandIds.length > 0 && retriedCommandIds.length === 0
          ? 'failed'
          : retriedCommandIds.length > 0
            ? 'retried'
            : 'released'
        await client.query(
          `UPDATE cloud_sessions
           SET status = $3, updated_at = $4
           WHERE tenant_id = $1 AND session_id = $2`,
          [
            lease.tenantId,
            lease.sessionId,
            action === 'failed' ? 'errored' : 'idle',
            nowIsoValue,
          ],
        )
        const org = await this.maybeOne(
          `SELECT org_id FROM cloud_orgs WHERE tenant_id = $1 OR org_id = $1 LIMIT 1`,
          [lease.tenantId],
          client,
        )
        if (org) {
          await this.recordAuditEventWithExecutor(client, {
            orgId: String(org.org_id),
            actorType: 'system',
            actorId: 'managed-work-reaper',
            eventType: 'managed_work.session_lease_reaped',
            targetType: 'session',
            targetId: lease.sessionId,
            metadata: {
              action,
              leasedBy: lease.leasedBy,
              retriedCommandIds,
              failedCommandIds,
            },
            createdAt: now,
          })
        }
        reaped.push({
          tenantId: lease.tenantId,
          sessionId: lease.sessionId,
          leaseToken: lease.leaseToken,
          leasedBy: lease.leasedBy,
          action,
          retriedCommandIds,
          failedCommandIds,
          reapedAt: nowIsoValue,
        })
      }
      return reaped
    })
  }

  async assertSessionCommandQueueQuota(input: { tenantId: string, quota?: CommandQueueQuota | null, now?: Date }) {
    await this.withTransaction(async (client) => {
      await assertPostgresCommandQueueQuota(client, input, this.quotaDeps)
    })
  }

  async enqueueSessionCommand(input: EnqueueCommandInput) {
    return this.withTransaction(async (client) => {
      await this.requireTenantUser(input.tenantId, input.userId, client)
      await this.requireSession(input.tenantId, input.sessionId, client, true)
      const payload = input.payload || {}
      const existing = await this.maybeOne(
        `SELECT * FROM cloud_session_commands WHERE command_id = $1`,
        [input.commandId],
        client,
      )
      if (existing) {
        const command = commandFromRow(existing)
        if (
          command.tenantId !== input.tenantId
          || command.userId !== input.userId
          || command.sessionId !== input.sessionId
          || command.kind !== input.kind
          || command.targetLeaseToken !== (input.targetLeaseToken ?? null)
          || stableJson(command.payload) !== stableJson(payload)
        ) {
          throw new Error(`Command id ${input.commandId} was reused with different content.`)
        }
        return command
      }
      const createdAt = nowIso(input.createdAt)
      await assertPostgresCommandEnqueueQuotas(client, {
        tenantId: input.tenantId,
        queueQuota: input.quota,
        usageQuotas: input.usageQuotas,
        now: new Date(createdAt),
      }, this.quotaDeps)
      const sequence = await this.incrementSessionCounter(
        client,
        input.tenantId,
        input.sessionId,
        'next_command_sequence',
      )
      const result = await client.query(
        `INSERT INTO cloud_session_commands (
          command_id, tenant_id, user_id, session_id, kind, payload,
          target_lease_token, created_sequence, created_at, status
         )
         VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, $9, 'pending')
         RETURNING *`,
        [
          input.commandId,
          input.tenantId,
          input.userId,
          input.sessionId,
          input.kind,
          JSON.stringify(payload),
          input.targetLeaseToken ?? null,
          sequence,
          createdAt,
        ],
      )
      return commandFromRow(result.rows[0])
    })
  }

  async claimNextSessionCommand(lease: WorkerLeaseRecord, now = new Date()) {
    return this.withTransaction(async (client) => {
      await this.assertCurrentLease(lease, client, now.getTime())
      const selected = await this.maybeOne(
        `SELECT * FROM cloud_session_commands
         WHERE tenant_id = $1
           AND session_id = $2
           AND (
             (status = 'pending'
                AND (available_at IS NULL OR available_at <= $4)
                AND (target_lease_token IS NULL OR target_lease_token = $3))
             OR (status = 'running' AND claimed_lease_token <> $3 AND target_lease_token IS NULL)
           )
         ORDER BY created_sequence
         FOR UPDATE SKIP LOCKED
         LIMIT 1`,
        [lease.tenantId, lease.sessionId, lease.leaseToken, now.toISOString()],
        client,
      )
      if (!selected) return null
      const result = await client.query(
        `UPDATE cloud_session_commands
         SET status = 'running',
             claimed_by = $2,
             claimed_lease_token = $3,
             attempt_count = attempt_count + 1,
             available_at = NULL,
             last_error_code = NULL,
             last_error_summary = NULL
         WHERE command_id = $1
         RETURNING *`,
        [String(selected.command_id), lease.leasedBy, lease.leaseToken],
      )
      return commandFromRow(result.rows[0])
    })
  }

  async ackSessionCommand(lease: WorkerLeaseRecord, commandId: string, now = new Date()) {
    return this.withTransaction(async (client) => {
      await this.assertCurrentLease(lease, client)
      const command = await this.requireCommand(commandId, client, true)
      if (command.status === 'acked') return command
      if (command.status !== 'running' || command.claimedLeaseToken !== lease.leaseToken) {
        throw new Error(`Command ${commandId} is not owned by this worker.`)
      }
      const result = await client.query(
        `UPDATE cloud_session_commands
         SET status = 'acked',
             acked_at = $2,
             error = NULL,
             last_error_code = NULL,
             last_error_summary = NULL
         WHERE command_id = $1
         RETURNING *`,
        [commandId, now.toISOString()],
      )
      return commandFromRow(result.rows[0])
    })
  }

  async failSessionCommand(lease: WorkerLeaseRecord, commandId: string, error: string) {
    return this.withTransaction(async (client) => {
      await this.assertCurrentLease(lease, client)
      const command = await this.requireCommand(commandId, client, true)
      if (command.status !== 'running' || command.claimedLeaseToken !== lease.leaseToken) {
        throw new Error(`Command ${commandId} is not owned by this worker.`)
      }
      const result = await client.query(
        `UPDATE cloud_session_commands
         SET status = 'failed',
             error = $2,
             last_error_code = 'execution_failed',
             last_error_summary = $3
         WHERE command_id = $1
         RETURNING *`,
        [commandId, error, redactOperationalText(error, 512, 'Command error')],
      )
      return commandFromRow(result.rows[0])
    })
  }

  async recordWorkerHeartbeat(input: {
    workerId: string
    role: WorkerRole
    activeSessionIds?: string[]
    now?: Date
  }) {
    const result = await this.pool.query(
      `INSERT INTO cloud_worker_heartbeats (worker_id, role, active_session_ids, last_seen_at)
       VALUES ($1, $2, $3::jsonb, $4)
       ON CONFLICT (worker_id) DO UPDATE
       SET role = EXCLUDED.role,
           active_session_ids = EXCLUDED.active_session_ids,
           last_seen_at = EXCLUDED.last_seen_at
       RETURNING *`,
      [
        input.workerId,
        input.role,
        JSON.stringify([...new Set(input.activeSessionIds || [])]),
        nowIso(input.now),
      ],
    )
    return heartbeatFromRow(result.rows[0])
  }

  async listWorkerHeartbeats() {
    const result = await this.pool.query(
      `SELECT * FROM cloud_worker_heartbeats ORDER BY worker_id`,
    )
    return result.rows.map(heartbeatFromRow)
  }

  async setSettingMetadata(input: {
    tenantId: string
    userId?: string | null
    key: string
    value: Record<string, unknown>
    updatedAt?: Date
  }) {
    await this.requireTenant(input.tenantId)
    if (input.userId) await this.requireTenantUser(input.tenantId, input.userId)
    const result = await this.pool.query(
      `INSERT INTO cloud_setting_metadata (
        tenant_id, user_scope, user_id, key, value, updated_at
       )
       VALUES ($1, $2, $3, $4, $5::jsonb, $6)
       ON CONFLICT (tenant_id, user_scope, key) DO UPDATE
       SET user_id = EXCLUDED.user_id,
           value = EXCLUDED.value,
           updated_at = EXCLUDED.updated_at
       RETURNING *`,
      [
        input.tenantId,
        input.userId || '',
        input.userId || null,
        input.key,
        JSON.stringify(input.value),
        nowIso(input.updatedAt),
      ],
    )
    return settingFromRow(result.rows[0])
  }

  async getSettingMetadata(tenantId: string, keyName: string, userId?: string | null) {
    await this.requireTenant(tenantId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_setting_metadata
       WHERE tenant_id = $1 AND user_scope = $2 AND key = $3`,
      [tenantId, userId || '', keyName],
    )
    return row ? settingFromRow(row) : null
  }

  async listSettingMetadata(tenantId: string, userId?: string | null) {
    await this.requireTenant(tenantId)
    if (userId) await this.requireTenantUser(tenantId, userId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_setting_metadata
       WHERE tenant_id = $1 AND user_scope = $2
       ORDER BY key`,
      [tenantId, userId || ''],
    )
    return result.rows.map(settingFromRow)
  }

  async createWorkflow(input: CreateWorkflowInput) {
    await this.requireTenantUser(input.tenantId, input.userId)
    const createdAt = nowIso(input.createdAt)
    const draft = input.draft
    await this.pool.query(
      `INSERT INTO cloud_workflows (
        tenant_id, workflow_id, user_id, title, instructions, agent_name,
        skill_names, tool_ids, status, project_directory, draft_session_id,
        triggers, created_at, updated_at, next_run_at, last_run_at,
        latest_run_id, latest_run_status, latest_run_session_id, latest_run_summary
       )
       VALUES (
        $1, $2, $3, $4, $5, $6,
        $7::jsonb, $8::jsonb, 'active', $9, $10,
        $11::jsonb, $12, $12, $13, NULL,
        NULL, NULL, NULL, NULL
       )
       ON CONFLICT (tenant_id, workflow_id) DO NOTHING`,
      [
        input.tenantId,
        input.workflowId,
        input.userId,
        draft.title,
        draft.instructions,
        draft.agentName,
        JSON.stringify(draft.skillNames || []),
        JSON.stringify(draft.toolIds || []),
        draft.projectDirectory || null,
        draft.draftSessionId || null,
        JSON.stringify(draft.triggers),
        createdAt,
        input.nextRunAt || null,
      ],
    )
    return workflowFromRow(await this.requireWorkflow(input.tenantId, input.workflowId))
  }

  async findWorkflow(workflowId: string) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflows
       WHERE workflow_id = $1
       ORDER BY updated_at DESC
       LIMIT 1`,
      [workflowId],
    )
    return row ? workflowFromRow(row) : null
  }

  async listWorkflows(tenantId: string, userId: string) {
    await this.requireTenantUser(tenantId, userId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_workflows
       WHERE tenant_id = $1 AND user_id = $2
       ORDER BY updated_at DESC, workflow_id`,
      [tenantId, userId],
    )
    return result.rows.map(workflowFromRow)
  }

  async getWorkflow(tenantId: string, userId: string, workflowId: string) {
    await this.requireTenantUser(tenantId, userId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflows
       WHERE tenant_id = $1 AND user_id = $2 AND workflow_id = $3`,
      [tenantId, userId, workflowId],
    )
    return row ? workflowFromRow(row) : null
  }

  async getWorkflowForTenant(tenantId: string, workflowId: string) {
    await this.requireTenant(tenantId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflows WHERE tenant_id = $1 AND workflow_id = $2`,
      [tenantId, workflowId],
    )
    return row ? workflowFromRow(row) : null
  }

  async updateWorkflowStatus(input: UpdateWorkflowStatusInput) {
    await this.requireTenantUser(input.tenantId, input.userId)
    const result = await this.pool.query(
      `UPDATE cloud_workflows
       SET status = $4,
           next_run_at = $5,
           updated_at = $6
       WHERE tenant_id = $1 AND user_id = $2 AND workflow_id = $3
       RETURNING *`,
      [
        input.tenantId,
        input.userId,
        input.workflowId,
        input.status,
        input.nextRunAt || null,
        nowIso(input.updatedAt),
      ],
    )
    return result.rows[0] ? workflowFromRow(result.rows[0]) : null
  }

  async listWorkflowRuns(tenantId: string, workflowId: string, limit = 25) {
    await this.requireWorkflow(tenantId, workflowId)
    const boundedLimit = Math.min(Math.max(1, limit), WORKFLOW_RUN_LIST_LIMIT)
    const result = await this.pool.query(
      `SELECT * FROM cloud_workflow_runs
       WHERE tenant_id = $1 AND workflow_id = $2
       ORDER BY created_at DESC, run_id
       LIMIT $3`,
      [tenantId, workflowId, boundedLimit],
    )
    return result.rows.map(workflowRunFromRow)
  }

  async createWorkflowRun(input: CreateWorkflowRunInput) {
    return this.withTransaction(async (client) => {
      await this.requireTenantUser(input.tenantId, input.userId, client)
      const workflow = workflowFromRow(await this.requireWorkflow(input.tenantId, input.workflowId, client, true))
      if (workflow.userId !== input.userId) throw new Error(`Unknown workflow ${input.workflowId}.`)
      const existing = await this.maybeOne(
        `SELECT * FROM cloud_workflow_runs WHERE tenant_id = $1 AND run_id = $2`,
        [input.tenantId, input.runId],
        client,
      )
      if (existing) return workflowRunFromRow(existing)
      this.assertWorkflowRunnable(workflow)
      const createdAt = nowIso(input.createdAt)
      await assertPostgresWorkflowRunQuota(client, {
        tenantId: input.tenantId,
        quota: input.quota,
        now: new Date(createdAt),
      }, this.quotaDeps)
      const claimedBy = input.claimedBy?.trim() || null
      const claimToken = claimedBy ? createWorkClaimToken(input.tenantId, input.runId, claimedBy) : null
      const leaseTtlMs = Math.max(1, Math.floor(input.leaseTtlMs ?? 30_000))
      const claimExpiresAt = claimToken ? new Date(new Date(createdAt).getTime() + leaseTtlMs).toISOString() : null
      const result = await client.query(
        `INSERT INTO cloud_workflow_runs (
          tenant_id, run_id, workflow_id, user_id, session_id, trigger_type,
          trigger_payload, status, title, summary, error, created_at, started_at, finished_at,
          claimed_by, claim_token, claim_expires_at, attempt_count, idempotency_key,
          checkpoint_version, last_error_code, last_error_summary
         )
         VALUES (
          $1, $2, $3, $4, NULL, $5, $6::jsonb, 'queued', $7, NULL, NULL, $8, NULL, NULL,
          $9, $10, $11, $12, NULL, 0, NULL, NULL
         )
         RETURNING *`,
        [
          input.tenantId,
          input.runId,
          input.workflowId,
          input.userId,
          input.triggerType,
          input.triggerPayload ? JSON.stringify(input.triggerPayload) : null,
          `Run ${workflow.title}`,
          createdAt,
          claimedBy,
          claimToken,
          claimExpiresAt,
          claimToken ? 1 : 0,
        ],
      )
      await client.query(
        `UPDATE cloud_workflows
         SET status = 'running',
             latest_run_id = $3,
             latest_run_status = 'queued',
             updated_at = $4
         WHERE tenant_id = $1 AND workflow_id = $2`,
        [input.tenantId, input.workflowId, input.runId, createdAt],
      )
      return workflowRunFromRow(result.rows[0])
    })
  }

  async claimDueWorkflowRun(input: ClaimDueWorkflowRunInput): Promise<ClaimedWorkflowRunRecord | null> {
    return this.withTransaction(async (client) => {
      const now = input.now || new Date()
      const claimedAt = now.toISOString()
      const claimedBy = input.claimedBy?.trim() || 'scheduler'
      const leaseTtlMs = Math.max(1, Math.floor(input.leaseTtlMs ?? 30_000))
      const claimExpiresAt = new Date(now.getTime() + leaseTtlMs).toISOString()
      const retryRow = await this.maybeOne(
        `SELECT runs.*, workflows.tenant_id AS workflow_tenant_id
         FROM cloud_workflow_runs runs
         JOIN cloud_workflows workflows
           ON workflows.tenant_id = runs.tenant_id
          AND workflows.workflow_id = runs.workflow_id
         WHERE runs.claim_token IS NULL
           AND (
             (runs.status = 'queued' AND runs.session_id IS NULL)
             OR (
               runs.status = 'running'
               AND runs.session_id IS NOT NULL
               AND NOT EXISTS (
                 SELECT 1
                 FROM cloud_session_commands commands
                 WHERE commands.tenant_id = runs.tenant_id
                   AND commands.session_id = runs.session_id
               )
             )
           )
           AND workflows.status = 'running'
         ORDER BY runs.created_at ASC, runs.run_id
         FOR UPDATE OF runs, workflows SKIP LOCKED
         LIMIT 1`,
        [],
        client,
      )
      if (retryRow) {
        const runId = String(retryRow.run_id)
        const tenantId = String(retryRow.tenant_id)
        const workflowId = String(retryRow.workflow_id)
        const retryStatus = String(retryRow.status)
        const retrySessionId = retryRow.session_id ? String(retryRow.session_id) : null
        const claimToken = createWorkClaimToken(tenantId, runId, claimedBy)
        const updatedRun = await client.query(
          `UPDATE cloud_workflow_runs
           SET claimed_by = $4,
               claim_token = $5,
               claim_expires_at = $6,
               attempt_count = attempt_count + 1,
               last_error_code = NULL,
               last_error_summary = NULL
           WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3
           RETURNING *`,
          [tenantId, workflowId, runId, claimedBy, claimToken, claimExpiresAt],
        )
        const updatedWorkflow = await client.query(
          `UPDATE cloud_workflows
           SET status = 'running',
               latest_run_id = $3,
               latest_run_status = $4,
               latest_run_session_id = $5,
               updated_at = $6
           WHERE tenant_id = $1 AND workflow_id = $2
           RETURNING *`,
          [tenantId, workflowId, runId, retryStatus, retrySessionId, claimedAt],
        )
        return {
          workflow: workflowFromRow(updatedWorkflow.rows[0]),
          run: workflowRunFromRow(updatedRun.rows[0]),
        }
      }
      const row = await this.maybeOne(
        `SELECT * FROM cloud_workflows
         WHERE status = 'active'
           AND next_run_at IS NOT NULL
           AND next_run_at <= $1
         ORDER BY next_run_at ASC, tenant_id, workflow_id
         FOR UPDATE SKIP LOCKED
         LIMIT 1`,
        [claimedAt],
        client,
      )
      if (!row) return null
      const workflow = workflowFromRow(row)
      await assertPostgresWorkflowRunQuota(client, {
        tenantId: workflow.tenantId,
        quota: input.quota,
        now,
      }, this.quotaDeps)
      const claimToken = createWorkClaimToken(workflow.tenantId, input.runId, claimedBy)
      const result = await client.query(
        `INSERT INTO cloud_workflow_runs (
          tenant_id, run_id, workflow_id, user_id, session_id, trigger_type,
          trigger_payload, status, title, summary, error, created_at, started_at, finished_at,
          claimed_by, claim_token, claim_expires_at, attempt_count, idempotency_key,
          checkpoint_version, last_error_code, last_error_summary
         )
         VALUES (
          $1, $2, $3, $4, NULL, 'schedule',
          $5::jsonb, 'queued', $6, NULL, NULL, $7, NULL, NULL,
          $8, $9, $10, 1, $11, 0, NULL, NULL
         )
         RETURNING *`,
        [
          workflow.tenantId,
          input.runId,
          workflow.id,
          workflow.userId,
          JSON.stringify({ source: 'schedule', scheduledFor: workflow.nextRunAt }),
          `Run ${workflow.title}`,
          claimedAt,
          claimedBy,
          claimToken,
          claimExpiresAt,
          `schedule:${workflow.id}:${workflow.nextRunAt}`,
        ],
      )
      const updatedWorkflow = await client.query(
        `UPDATE cloud_workflows
         SET status = 'running',
             latest_run_id = $3,
             latest_run_status = 'queued',
             updated_at = $4
         WHERE tenant_id = $1 AND workflow_id = $2
         RETURNING *`,
        [workflow.tenantId, workflow.id, input.runId, claimedAt],
      )
      return {
        workflow: workflowFromRow(updatedWorkflow.rows[0]),
        run: workflowRunFromRow(result.rows[0]),
      }
    })
  }

  async reapExpiredWorkflowClaims(input: ReapExpiredWorkflowClaimsInput = {}): Promise<ReapedWorkflowClaimRecord[]> {
    const now = input.now || new Date()
    const nowIsoValue = now.toISOString()
    const maxAttempts = Math.max(1, Math.floor(input.maxAttempts ?? 3))
    const limit = Math.max(1, Math.min(1_000, Math.floor(input.limit ?? 100)))
    return this.withTransaction(async (client) => {
      const expired = await client.query(
        `SELECT *
         FROM cloud_workflow_runs
         WHERE claim_token IS NOT NULL
           AND claim_expires_at IS NOT NULL
           AND claim_expires_at <= $1
           AND (
             (status = 'queued' AND session_id IS NULL)
             OR (
               status = 'running'
               AND session_id IS NOT NULL
               AND NOT EXISTS (
                 SELECT 1
                 FROM cloud_session_commands commands
                 WHERE commands.tenant_id = cloud_workflow_runs.tenant_id
                   AND commands.session_id = cloud_workflow_runs.session_id
               )
             )
           )
         ORDER BY claim_expires_at ASC, tenant_id, workflow_id, run_id
         LIMIT $2
         FOR UPDATE SKIP LOCKED`,
        [nowIsoValue, limit],
      )
      const reaped: ReapedWorkflowClaimRecord[] = []
      for (const row of expired.rows) {
        const run = workflowRunFromRow(row)
        await this.requireWorkflow(run.tenantId, run.workflowId, client, true)
        const claimToken = run.claimToken
        if (!claimToken) continue
        const claimedBy = run.claimedBy || 'unknown'
        const action: ReapedWorkflowClaimRecord['action'] = run.attemptCount >= maxAttempts ? 'failed' : 'retried'
        if (action === 'failed') {
          const summary = 'Workflow run claim expired after the maximum retry attempts.'
          await client.query(
            `UPDATE cloud_workflow_runs
             SET status = 'failed',
                 summary = $4,
                 error = $4,
                 finished_at = $5,
                 claimed_by = NULL,
                 claim_token = NULL,
                 claim_expires_at = NULL,
                 last_error_code = 'claim_expired_max_attempts',
                 last_error_summary = $4
             WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3`,
            [run.tenantId, run.workflowId, run.id, summary, nowIsoValue],
          )
          await client.query(
            `UPDATE cloud_workflows
             SET status = 'failed',
                 latest_run_id = $3,
                 latest_run_status = 'failed',
                 latest_run_summary = $4,
                 next_run_at = NULL,
                 updated_at = $5
             WHERE tenant_id = $1 AND workflow_id = $2`,
            [run.tenantId, run.workflowId, run.id, summary, nowIsoValue],
          )
        } else {
          await client.query(
            `UPDATE cloud_workflow_runs
             SET claimed_by = NULL,
                 claim_token = NULL,
                 claim_expires_at = NULL,
                 last_error_code = 'claim_expired',
                 last_error_summary = $4
             WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3`,
            [
              run.tenantId,
              run.workflowId,
              run.id,
              run.sessionId
                ? 'Workflow run claim expired before command enqueue.'
                : 'Workflow run claim expired before session attachment.',
            ],
          )
          await client.query(
            `UPDATE cloud_workflows
             SET status = 'running',
                 latest_run_id = $3,
                 latest_run_status = $4,
                 latest_run_session_id = $5,
                 updated_at = $6
             WHERE tenant_id = $1 AND workflow_id = $2`,
            [run.tenantId, run.workflowId, run.id, run.status, run.sessionId, nowIsoValue],
          )
        }
        reaped.push({
          tenantId: run.tenantId,
          workflowId: run.workflowId,
          runId: run.id,
          claimToken,
          claimedBy,
          action,
          reapedAt: nowIsoValue,
        })
      }
      return reaped
    })
  }

  async attachWorkflowRunSession(input: AttachWorkflowRunSessionInput) {
    return this.withTransaction(async (client) => {
      await this.requireWorkflow(input.tenantId, input.workflowId, client, true)
      const runRow = await this.maybeOne(
        `SELECT * FROM cloud_workflow_runs
         WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3
         FOR UPDATE`,
        [input.tenantId, input.workflowId, input.runId],
        client,
      )
      if (!runRow) return null
      const current = workflowRunFromRow(runRow)
      if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') {
        throw new Error('Workflow run is not attachable.')
      }
      if (current.status !== 'queued' && !(current.status === 'running' && current.sessionId === input.sessionId)) {
        throw new Error('Workflow run is not attachable.')
      }
      if (current.sessionId && current.sessionId !== input.sessionId) throw new Error('Workflow run is already attached to another session.')
      if (current.claimToken) {
        if (current.claimToken !== (input.claimToken ?? null)) throw new Error('Workflow run claim is stale.')
        if (current.claimExpiresAt && Date.parse(current.claimExpiresAt) <= Date.now()) {
          throw new Error('Workflow run claim is stale.')
        }
      } else if (input.claimToken) {
        throw new Error('Workflow run claim is stale.')
      }
      const startedAt = nowIso(input.startedAt)
      const result = await client.query(
        `UPDATE cloud_workflow_runs
         SET session_id = $4,
             status = 'running',
             started_at = COALESCE(started_at, $5),
             claimed_by = NULL,
             claim_token = NULL,
             claim_expires_at = NULL
         WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3
         RETURNING *`,
        [input.tenantId, input.workflowId, input.runId, input.sessionId, startedAt],
      )
      if (!result.rows[0]) return null
      await client.query(
        `UPDATE cloud_workflows
         SET status = 'running',
             latest_run_id = $3,
             latest_run_status = 'running',
             latest_run_session_id = $4,
             updated_at = $5
         WHERE tenant_id = $1 AND workflow_id = $2`,
        [input.tenantId, input.workflowId, input.runId, input.sessionId, startedAt],
      )
      return workflowRunFromRow(result.rows[0])
    })
  }

  async completeWorkflowRun(input: CompleteWorkflowRunInput) {
    return this.finishWorkflowRun({
      tenantId: input.tenantId,
      workflowId: input.workflowId,
      runId: input.runId,
      status: 'completed',
      summary: input.summary,
      error: null,
      nextStatus: input.nextStatus,
      nextRunAt: input.nextRunAt,
      leaseToken: input.leaseToken,
      finishedAt: input.finishedAt,
    })
  }

  async failWorkflowRun(input: FailWorkflowRunInput) {
    return this.finishWorkflowRun({
      tenantId: input.tenantId,
      workflowId: input.workflowId,
      runId: input.runId,
      status: 'failed',
      summary: input.error,
      error: input.error,
      nextStatus: input.nextStatus,
      nextRunAt: input.nextRunAt,
      leaseToken: input.leaseToken,
      finishedAt: input.finishedAt,
    })
  }

  async getWorkflowRun(tenantId: string, runId: string) {
    await this.requireTenant(tenantId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflow_runs WHERE tenant_id = $1 AND run_id = $2`,
      [tenantId, runId],
    )
    return row ? workflowRunFromRow(row) : null
  }

  async getWorkflowRunBySession(tenantId: string, sessionId: string) {
    await this.requireTenant(tenantId)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflow_runs
       WHERE tenant_id = $1 AND session_id = $2
       ORDER BY created_at DESC
       LIMIT 1`,
      [tenantId, sessionId],
    )
    return row ? workflowRunFromRow(row) : null
  }

  async listThreadTags(tenantId: string) {
    await this.requireTenant(tenantId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_thread_tags
       WHERE tenant_id = $1
       ORDER BY lower(name), tag_id`,
      [tenantId],
    )
    return result.rows.map(threadTagFromRow)
  }

  async createThreadTag(input: CreateThreadTagInput) {
    await this.requireTenant(input.tenantId)
    const name = normalizeText(input.name, THREAD_TAG_NAME_MAX_LENGTH, 'Tag name')
    const color = normalizeTagColor(input.color)
    const createdAt = nowIso(input.createdAt)
    const existing = await this.maybeOne(
      `SELECT * FROM cloud_thread_tags WHERE tenant_id = $1 AND tag_id = $2`,
      [input.tenantId, input.tagId],
    )
    if (existing) {
      const tag = threadTagFromRow(existing)
      if (tag.name !== name || tag.color !== color) {
        throw new Error(`Tag id ${input.tagId} was reused with different content.`)
      }
      return tag
    }
    const result = await this.pool.query(
      `INSERT INTO cloud_thread_tags (tenant_id, tag_id, name, color, created_at, updated_at)
       VALUES ($1, $2, $3, $4, $5, $5)
       RETURNING *`,
      [input.tenantId, input.tagId, name, color, createdAt],
    )
    return threadTagFromRow(result.rows[0])
  }

  async updateThreadTag(input: UpdateThreadTagInput) {
    await this.requireTenant(input.tenantId)
    const name = normalizeOptionalText(input.name, THREAD_TAG_NAME_MAX_LENGTH, 'Tag name')
    const color = input.color === undefined ? undefined : normalizeTagColor(input.color)
    const result = await this.pool.query(
      `UPDATE cloud_thread_tags
       SET name = COALESCE($3, name),
           color = COALESCE($4, color),
           updated_at = $5
       WHERE tenant_id = $1 AND tag_id = $2
       RETURNING *`,
      [input.tenantId, input.tagId, name ?? null, color ?? null, nowIso(input.updatedAt)],
    )
    return result.rows[0] ? threadTagFromRow(result.rows[0]) : null
  }

  async deleteThreadTag(tenantId: string, tagId: string) {
    await this.requireTenant(tenantId)
    const result = await this.pool.query(
      `DELETE FROM cloud_thread_tags WHERE tenant_id = $1 AND tag_id = $2`,
      [tenantId, tagId],
    ) as QueryResult & { rowCount?: number }
    return Number(result.rowCount || 0) > 0
  }

  async applyThreadTags(input: ThreadTagLinkInput) {
    await this.withTransaction(async (client) => {
      await this.requireTenant(input.tenantId, client)
      const sessionIds = normalizeIdList(input.sessionIds, 'sessionIds', THREAD_BULK_MAX_SESSION_IDS)
      const tagIds = normalizeIdList(input.tagIds, 'tagIds', THREAD_FILTER_MAX_VALUES)
      for (const sessionId of sessionIds) await this.requireSession(input.tenantId, sessionId, client)
      for (const tagId of tagIds) await this.requireThreadTag(input.tenantId, tagId, client)
      const createdAt = nowIso(input.createdAt)
      for (const sessionId of sessionIds) {
        for (const tagId of tagIds) {
          await client.query(
            `INSERT INTO cloud_thread_tag_links (tenant_id, session_id, tag_id, created_at)
             VALUES ($1, $2, $3, $4)
             ON CONFLICT (tenant_id, session_id, tag_id) DO NOTHING`,
            [input.tenantId, sessionId, tagId, createdAt],
          )
        }
      }
    })
  }

  async removeThreadTags(input: ThreadTagLinkInput) {
    await this.withTransaction(async (client) => {
      await this.requireTenant(input.tenantId, client)
      const sessionIds = normalizeIdList(input.sessionIds, 'sessionIds', THREAD_BULK_MAX_SESSION_IDS)
      const tagIds = normalizeIdList(input.tagIds, 'tagIds', THREAD_FILTER_MAX_VALUES)
      for (const sessionId of sessionIds) await this.requireSession(input.tenantId, sessionId, client)
      for (const tagId of tagIds) await this.requireThreadTag(input.tenantId, tagId, client)
      if (sessionIds.length === 0 || tagIds.length === 0) return
      await client.query(
        `DELETE FROM cloud_thread_tag_links
         WHERE tenant_id = $1
           AND session_id = ANY($2::text[])
           AND tag_id = ANY($3::text[])`,
        [input.tenantId, sessionIds, tagIds],
      )
    })
  }

  async listThreadSmartFilters(tenantId: string) {
    await this.requireTenant(tenantId)
    const result = await this.pool.query(
      `SELECT * FROM cloud_thread_smart_filters
       WHERE tenant_id = $1
       ORDER BY lower(name), filter_id`,
      [tenantId],
    )
    return result.rows.map(threadSmartFilterFromRow)
  }

  async createThreadSmartFilter(input: CreateThreadSmartFilterInput) {
    await this.requireTenant(input.tenantId)
    const name = normalizeText(input.name, THREAD_SMART_FILTER_NAME_MAX_LENGTH, 'Smart filter name')
    const query = normalizeThreadQuery(input.query)
    const createdAt = nowIso(input.createdAt)
    const existing = await this.maybeOne(
      `SELECT * FROM cloud_thread_smart_filters WHERE tenant_id = $1 AND filter_id = $2`,
      [input.tenantId, input.filterId],
    )
    if (existing) {
      const filter = threadSmartFilterFromRow(existing)
      if (filter.name !== name || stableJson(filter.query) !== stableJson(query)) {
        throw new Error(`Smart filter id ${input.filterId} was reused with different content.`)
      }
      return filter
    }
    const result = await this.pool.query(
      `INSERT INTO cloud_thread_smart_filters (tenant_id, filter_id, name, query, created_at, updated_at)
       VALUES ($1, $2, $3, $4::jsonb, $5, $5)
       RETURNING *`,
      [input.tenantId, input.filterId, name, JSON.stringify(query), createdAt],
    )
    return threadSmartFilterFromRow(result.rows[0])
  }

  async updateThreadSmartFilter(input: UpdateThreadSmartFilterInput) {
    await this.requireTenant(input.tenantId)
    const name = normalizeOptionalText(input.name, THREAD_SMART_FILTER_NAME_MAX_LENGTH, 'Smart filter name')
    const query = input.query === undefined ? undefined : normalizeThreadQuery(input.query)
    const result = await this.pool.query(
      `UPDATE cloud_thread_smart_filters
       SET name = COALESCE($3, name),
           query = COALESCE($4::jsonb, query),
           updated_at = $5
       WHERE tenant_id = $1 AND filter_id = $2
       RETURNING *`,
      [
        input.tenantId,
        input.filterId,
        name ?? null,
        query === undefined ? null : JSON.stringify(query),
        nowIso(input.updatedAt),
      ],
    )
    return result.rows[0] ? threadSmartFilterFromRow(result.rows[0]) : null
  }

  async deleteThreadSmartFilter(tenantId: string, filterId: string) {
    await this.requireTenant(tenantId)
    const result = await this.pool.query(
      `DELETE FROM cloud_thread_smart_filters WHERE tenant_id = $1 AND filter_id = $2`,
      [tenantId, filterId],
    ) as QueryResult & { rowCount?: number }
    return Number(result.rowCount || 0) > 0
  }

  async listThreadMetadata(input: {
    tenantId: string
    userId: string
    tagIds?: string[]
    limit?: number
  }): Promise<ThreadMetadataRecord[]> {
    await this.requireTenantUser(input.tenantId, input.userId)
    const tagIds = input.tagIds
      ? normalizeIdList(input.tagIds, 'tagIds', THREAD_FILTER_MAX_VALUES)
      : []
    const limit = Number.isInteger(input.limit) && input.limit && input.limit > 0
      ? Math.min(input.limit, THREAD_BULK_MAX_SESSION_IDS)
      : THREAD_BULK_MAX_SESSION_IDS
    const result = tagIds.length > 0
      ? await this.pool.query(
        `SELECT * FROM cloud_sessions s
         WHERE s.tenant_id = $1
           AND s.user_id = $2
           AND EXISTS (
             SELECT 1 FROM cloud_thread_tag_links link
             WHERE link.tenant_id = s.tenant_id
               AND link.session_id = s.session_id
               AND link.tag_id = ANY($3::text[])
           )
         ORDER BY s.updated_at DESC, s.session_id
         LIMIT $4`,
        [input.tenantId, input.userId, tagIds, limit],
      )
      : await this.pool.query(
        `SELECT * FROM cloud_sessions
         WHERE tenant_id = $1 AND user_id = $2
         ORDER BY updated_at DESC, session_id
         LIMIT $3`,
        [input.tenantId, input.userId, limit],
      )
    return Promise.all(result.rows.map(async (row) => {
      const session = sessionFromRow(row)
      return {
        tenantId: session.tenantId,
        userId: session.userId,
        sessionId: session.sessionId,
        title: session.title,
        profileName: session.profileName,
        status: session.status,
        createdAt: session.createdAt,
        updatedAt: session.updatedAt,
        tags: await this.listThreadTagsForSession(session.tenantId, session.sessionId),
      }
    }))
  }

  async recordSchemaMigration(id: string, appliedAt = new Date()) {
    await this.pool.query(
      `INSERT INTO cloud_schema_migrations (id, applied_at)
       VALUES ($1, $2)
       ON CONFLICT (id) DO NOTHING`,
      [id, appliedAt.toISOString()],
    )
    const row = await this.one(
      `SELECT * FROM cloud_schema_migrations WHERE id = $1`,
      [id],
    )
    return migrationFromRow(row)
  }

  async listSchemaMigrations() {
    const result = await this.pool.query(
      `SELECT * FROM cloud_schema_migrations ORDER BY applied_at, id`,
    )
    return result.rows.map(migrationFromRow)
  }

  async claimRequest(input: {
    source: string
    nowMs: number
    windowMs: number
    limit: number
  }) {
    const result = await this.pool.query(
      `INSERT INTO cloud_webhook_rate_limits (source, window_started_at_ms, count)
       VALUES ($1, $2, 1)
       ON CONFLICT (source) DO UPDATE
       SET window_started_at_ms = CASE
             WHEN $2 - cloud_webhook_rate_limits.window_started_at_ms > $3 THEN $2
             ELSE cloud_webhook_rate_limits.window_started_at_ms
           END,
           count = CASE
             WHEN $2 - cloud_webhook_rate_limits.window_started_at_ms > $3 THEN 1
             ELSE cloud_webhook_rate_limits.count + 1
           END
       RETURNING count`,
      [input.source, input.nowMs, input.windowMs],
    )
    return numberValue(result.rows[0]?.count) <= input.limit
  }

  async checkAuthBackoff(input: { scope: string, nowMs: number }) {
    const row = await this.maybeOne(
      `SELECT blocked_until_ms FROM cloud_webhook_auth_failures WHERE scope = $1`,
      [input.scope],
    )
    return !row || numberValue(row.blocked_until_ms) <= input.nowMs
  }

  async recordAuthFailure(input: {
    scope: string
    source: string
    nowMs: number
    windowMs: number
    limit: number
    backoffMs: number
  }) {
    const blockedUntil = input.nowMs + input.backoffMs
    const result = await this.pool.query(
      `INSERT INTO cloud_webhook_auth_failures (
        scope, source, auth_window_started_at_ms, auth_failure_count, blocked_until_ms
       )
       VALUES ($1, $2, $3, 1, CASE WHEN $5 <= 1 THEN $6 ELSE 0 END)
       ON CONFLICT (scope) DO UPDATE
       SET source = EXCLUDED.source,
           auth_window_started_at_ms = CASE
             WHEN $3 - cloud_webhook_auth_failures.auth_window_started_at_ms > $4 THEN $3
             ELSE cloud_webhook_auth_failures.auth_window_started_at_ms
           END,
           auth_failure_count = CASE
             WHEN $3 - cloud_webhook_auth_failures.auth_window_started_at_ms > $4 THEN 1
             ELSE cloud_webhook_auth_failures.auth_failure_count + 1
           END,
           blocked_until_ms = CASE
             WHEN (
               CASE
                 WHEN $3 - cloud_webhook_auth_failures.auth_window_started_at_ms > $4 THEN 1
                 ELSE cloud_webhook_auth_failures.auth_failure_count + 1
               END
             ) >= $5 THEN GREATEST(cloud_webhook_auth_failures.blocked_until_ms, $6)
             ELSE cloud_webhook_auth_failures.blocked_until_ms
           END
       RETURNING *`,
      [input.scope, input.source, input.nowMs, input.windowMs, input.limit, blockedUntil],
    )
    return webhookAuthFailureFromRow(result.rows[0])
  }

  async claimSignature(input: {
    key: string
    nowMs: number
    windowMs: number
    cacheLimit: number
  }): Promise<WorkflowWebhookReplayClaim | null> {
    const claimed = await this.withTransaction(async (client) => {
      await client.query(
        `DELETE FROM cloud_webhook_replay_claims WHERE $1 - seen_at_ms > $2`,
        [input.nowMs, input.windowMs],
      )
      const result = await client.query(
        `INSERT INTO cloud_webhook_replay_claims (replay_key, seen_at_ms, status)
         VALUES ($1, $2, 'pending')
         ON CONFLICT (replay_key) DO NOTHING
         RETURNING replay_key`,
        [input.key, input.nowMs],
      )
      await client.query(
        `DELETE FROM cloud_webhook_replay_claims
         WHERE replay_key IN (
           SELECT replay_key
           FROM cloud_webhook_replay_claims
           ORDER BY seen_at_ms ASC
           OFFSET $1
         )`,
        [input.cacheLimit],
      )
      return Boolean(result.rows[0])
    })
    if (!claimed) return null
    let active = true
    return {
      accept: async () => {
        if (!active) return
        active = false
        await this.pool.query(
          `UPDATE cloud_webhook_replay_claims
           SET status = 'accepted'
           WHERE replay_key = $1`,
          [input.key],
        )
      },
      release: async () => {
        if (!active) return
        active = false
        await this.pool.query(
          `DELETE FROM cloud_webhook_replay_claims
           WHERE replay_key = $1 AND status = 'pending'`,
          [input.key],
        )
      },
    }
  }

  async clear() {
    await this.pool.query('DELETE FROM cloud_managed_worker_heartbeats')
    await this.pool.query('DELETE FROM cloud_worker_credentials')
    await this.pool.query('DELETE FROM cloud_managed_workers')
    await this.pool.query('DELETE FROM cloud_worker_pools')
    await this.pool.query('DELETE FROM cloud_subscriptions')
    await this.pool.query('DELETE FROM cloud_auth_failures')
    await this.pool.query('DELETE FROM cloud_rate_limits')
    await this.pool.query('DELETE FROM cloud_quota_locks')
    await this.pool.query('DELETE FROM cloud_usage_counters')
    await this.pool.query('DELETE FROM cloud_usage_events')
    await this.pool.query('DELETE FROM cloud_webhook_replay_claims')
    await this.pool.query('DELETE FROM cloud_webhook_auth_failures')
    await this.pool.query('DELETE FROM cloud_webhook_rate_limits')
  }

  private async recordAuditEventWithExecutor(
    executor: PgExecutor,
    input: RecordAuditEventInput,
  ): Promise<AuditEventRecord> {
    const eventId = input.eventId || randomRecordId('audit')
    const result = await executor.query(
      `INSERT INTO cloud_audit_events (
        event_id, org_id, account_id, actor_type, actor_id,
        event_type, target_type, target_id, metadata, created_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10)
       ON CONFLICT (event_id) DO NOTHING
       RETURNING *`,
      [
        eventId,
        input.orgId,
        input.accountId || null,
        input.actorType,
        input.actorId || null,
        input.eventType,
        input.targetType || null,
        input.targetId || null,
        JSON.stringify(redactAuditMetadata(input.metadata)),
        nowIso(input.createdAt),
      ],
    )
    if (result.rows[0]) return auditEventFromRow(result.rows[0])
    const existing = await this.maybeOne(`SELECT * FROM cloud_audit_events WHERE event_id = $1`, [eventId], executor)
    if (!existing) throw new Error(`Audit event ${eventId} was not recorded.`)
    return auditEventFromRow(existing)
  }

  private async recordUsageEventWithExecutor(
    executor: PgExecutor,
    input: RecordUsageEventInput,
  ): Promise<UsageEventRecord> {
    const eventId = input.eventId || randomRecordId('usage')
    const result = await executor.query(
      `INSERT INTO cloud_usage_events (
        event_id, org_id, account_id, event_type, quantity, unit, metadata, created_at
       )
       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8)
       ON CONFLICT (event_id) DO NOTHING
       RETURNING *`,
      [
        eventId,
        input.orgId,
        input.accountId || null,
        input.eventType,
        normalizePositiveInteger(input.quantity || 1, 'Usage quantity'),
        input.unit || 'count',
        JSON.stringify(redactAuditMetadata(input.metadata)),
        nowIso(input.createdAt),
      ],
    )
    if (result.rows[0]) return usageEventFromRow(result.rows[0])
    const existing = await this.maybeOne(`SELECT * FROM cloud_usage_events WHERE event_id = $1`, [eventId], executor)
    if (!existing) throw new Error(`Usage event ${eventId} was not recorded.`)
    return usageEventFromRow(existing)
  }

  private async consumeUsageQuotaWithExecutor(
    executor: PgExecutor,
    input: ConsumeUsageQuotaInput,
  ) {
    const limit = normalizePositiveInteger(input.limit, 'Quota limit')
    const quantity = normalizePositiveInteger(input.quantity || 1, 'Quota quantity')
    const windowMs = normalizePositiveInteger(input.windowMs, 'Quota window')
    const now = input.now || new Date()
    const nowMs = now.getTime()
    const startedAtMs = windowStart(nowMs, windowMs)
    await executor.query(
      `INSERT INTO cloud_usage_counters (org_id, quota_key, window_started_at_ms, quantity)
       VALUES ($1, $2, $3, 0)
       ON CONFLICT (org_id, quota_key) DO NOTHING`,
      [input.orgId, input.quotaKey, startedAtMs],
    )
    const row = await this.one(
      `SELECT * FROM cloud_usage_counters
       WHERE org_id = $1 AND quota_key = $2
       FOR UPDATE`,
      [input.orgId, input.quotaKey],
      executor,
    )
    const currentStartedAtMs = numberValue(row.window_started_at_ms)
    const currentQuantity = currentStartedAtMs === startedAtMs ? numberValue(row.quantity) : 0
    const nextQuantity = currentQuantity + quantity
    const resetMs = retryAfterMs(nowMs, startedAtMs, windowMs)
    const resetAt = new Date(nowMs + resetMs).toISOString()
    if (nextQuantity > limit) {
      return {
        allowed: false,
        orgId: input.orgId,
        quotaKey: input.quotaKey,
        limit,
        used: currentQuantity,
        remaining: Math.max(0, limit - currentQuantity),
        resetAt,
        retryAfterMs: resetMs,
        policyCode: input.policyCode,
      }
    }
    await executor.query(
      `UPDATE cloud_usage_counters
       SET window_started_at_ms = $3, quantity = $4
       WHERE org_id = $1 AND quota_key = $2`,
      [input.orgId, input.quotaKey, startedAtMs, nextQuantity],
    )
    return {
      allowed: true,
      orgId: input.orgId,
      quotaKey: input.quotaKey,
      limit,
      used: nextQuantity,
      remaining: Math.max(0, limit - nextQuantity),
      resetAt,
      retryAfterMs: resetMs,
      policyCode: input.policyCode,
    }
  }

  private async lockQuota(
    executor: PgExecutor,
    orgId: string,
    quotaKey: string,
    now = new Date(),
  ) {
    await executor.query(
      `INSERT INTO cloud_quota_locks (org_id, quota_key, updated_at)
       VALUES ($1, $2, $3)
       ON CONFLICT (org_id, quota_key) DO UPDATE
       SET updated_at = EXCLUDED.updated_at`,
      [orgId, quotaKey, now.toISOString()],
    )
    await this.one(
      `SELECT * FROM cloud_quota_locks WHERE org_id = $1 AND quota_key = $2 FOR UPDATE`,
      [orgId, quotaKey],
      executor,
    )
  }

  private async requireTenant(tenantId: string, executor: PgExecutor = this.pool) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_tenants WHERE tenant_id = $1`,
      [tenantId],
      executor,
    )
    if (!row) throw new Error(`Unknown tenant ${tenantId}.`)
    return tenantFromRow(row)
  }

  private async requireOrg(orgId: string, executor: PgExecutor = this.pool) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_orgs WHERE org_id = $1`,
      [orgId],
      executor,
    )
    if (!row) throw new Error(`Unknown org ${orgId}.`)
    return orgFromRow(row)
  }

  private async requireTenantUser(
    tenantId: string,
    userId: string,
    executor: PgExecutor = this.pool,
  ) {
    await this.requireTenant(tenantId, executor)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_users WHERE tenant_id = $1 AND user_id = $2`,
      [tenantId, userId],
      executor,
    )
    if (!row) throw new Error(`User ${userId} does not belong to tenant ${tenantId}.`)
    return userFromRow(row)
  }

  private async requireSession(
    tenantId: string,
    sessionId: string,
    executor: PgExecutor = this.pool,
    forUpdate = false,
  ) {
    await this.requireTenant(tenantId, executor)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_sessions
       WHERE tenant_id = $1 AND session_id = $2${forUpdate ? ' FOR UPDATE' : ''}`,
      [tenantId, sessionId],
      executor,
    )
    if (!row) throw new Error(`Unknown session ${sessionId}.`)
    return row
  }

  private async requireWorkflow(
    tenantId: string,
    workflowId: string,
    executor: PgExecutor = this.pool,
    forUpdate = false,
  ) {
    await this.requireTenant(tenantId, executor)
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workflows
       WHERE tenant_id = $1 AND workflow_id = $2${forUpdate ? ' FOR UPDATE' : ''}`,
      [tenantId, workflowId],
      executor,
    )
    if (!row) throw new Error(`Unknown workflow ${workflowId}.`)
    return row
  }

  private assertWorkflowRunnable(workflow: CloudWorkflowRecord) {
    if (workflow.status === 'archived') throw new Error('Archived workflows cannot run.')
    if (workflow.status === 'paused') throw new Error('Paused workflows cannot run.')
    if (workflow.status === 'running') throw new Error('Workflow is already running.')
  }

  private async finishWorkflowRun(input: {
    tenantId: string
    workflowId: string
    runId: string
    status: Extract<WorkflowRunStatus, 'completed' | 'failed'>
    summary: string | null
    error: string | null
    nextStatus: WorkflowStatus
    nextRunAt: string | null
    leaseToken?: string | null
    finishedAt?: Date
  }): Promise<CloudWorkflowRunRecord | null> {
    return this.withTransaction(async (client) => {
      await this.requireWorkflow(input.tenantId, input.workflowId, client, true)
      const runRow = await this.maybeOne(
        `SELECT * FROM cloud_workflow_runs
         WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3
         FOR UPDATE`,
        [input.tenantId, input.workflowId, input.runId],
        client,
      )
      if (!runRow) return null
      const current = workflowRunFromRow(runRow)
      if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') {
        return current
      }
      if (input.leaseToken !== undefined) {
        if (!current.sessionId) throw new Error('Workflow run has no execution session to fence.')
        await this.assertLeaseTokenIfPresent(input.tenantId, current.sessionId, input.leaseToken, client)
      }
      const finishedAt = nowIso(input.finishedAt)
      const result = await client.query(
        `UPDATE cloud_workflow_runs
         SET status = $4,
             summary = $5,
             error = $6,
             finished_at = $7
         WHERE tenant_id = $1 AND workflow_id = $2 AND run_id = $3
         RETURNING *`,
        [
          input.tenantId,
          input.workflowId,
          input.runId,
          input.status,
          input.summary,
          input.error,
          finishedAt,
        ],
      )
      await client.query(
        `UPDATE cloud_workflows
         SET status = $4,
             latest_run_id = $3,
             latest_run_status = $5,
             latest_run_summary = $6,
             last_run_at = CASE WHEN $5 = 'completed' THEN $7 ELSE last_run_at END,
             next_run_at = $8,
             updated_at = $7
         WHERE tenant_id = $1 AND workflow_id = $2`,
        [
          input.tenantId,
          input.workflowId,
          input.runId,
          input.nextStatus,
          input.status,
          input.summary,
          finishedAt,
          input.nextRunAt,
        ],
      )
      return workflowRunFromRow(result.rows[0])
    })
  }

  private async requireThreadTag(
    tenantId: string,
    tagId: string,
    executor: PgExecutor = this.pool,
  ) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_thread_tags WHERE tenant_id = $1 AND tag_id = $2`,
      [tenantId, tagId],
      executor,
    )
    if (!row) throw new Error(`Unknown thread tag ${tagId}.`)
    return threadTagFromRow(row)
  }

  private async listThreadTagsForSession(
    tenantId: string,
    sessionId: string,
    executor: PgExecutor = this.pool,
  ) {
    const result = await executor.query(
      `SELECT tag.*
       FROM cloud_thread_tags tag
       JOIN cloud_thread_tag_links link
         ON link.tenant_id = tag.tenant_id
        AND link.tag_id = tag.tag_id
       WHERE link.tenant_id = $1 AND link.session_id = $2
       ORDER BY lower(tag.name), tag.tag_id`,
      [tenantId, sessionId],
    )
    return result.rows.map(threadTagFromRow)
  }

  private async requireCommand(commandId: string, executor: PgExecutor, forUpdate = false) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_session_commands
       WHERE command_id = $1${forUpdate ? ' FOR UPDATE' : ''}`,
      [commandId],
      executor,
    )
    if (!row) throw new Error(`Unknown command ${commandId}.`)
    return commandFromRow(row)
  }

  private async assertCurrentLease(lease: WorkerLeaseRecord, executor: PgExecutor, nowMs = Date.now()) {
    const current = await this.getLease(lease.tenantId, lease.sessionId, executor, true)
    if (!current || current.leaseToken !== lease.leaseToken || current.leaseExpiresAt <= nowMs) {
      throw new Error('Worker lease is stale.')
    }
    return current
  }

  private async assertLeaseTokenIfPresent(
    tenantId: string,
    sessionId: string,
    leaseToken: string | null | undefined,
    executor: PgExecutor,
  ) {
    if (leaseToken === undefined) return
    const current = await this.getLease(tenantId, sessionId, executor, true)
    if (!current || current.leaseToken !== leaseToken || current.leaseExpiresAt <= Date.now()) {
      throw new Error('Worker lease is stale.')
    }
  }

  private async getLease(
    tenantId: string,
    sessionId: string,
    executor: PgExecutor,
    forUpdate = false,
  ) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_worker_leases
       WHERE tenant_id = $1 AND session_id = $2${forUpdate ? ' FOR UPDATE' : ''}`,
      [tenantId, sessionId],
      executor,
    )
    return row ? leaseFromRow(row) : null
  }

  private async findEvent(
    tenantId: string,
    sessionId: string,
    eventId: string,
    executor: PgExecutor,
  ) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_session_events
       WHERE tenant_id = $1 AND session_id = $2 AND event_id = $3`,
      [tenantId, sessionId, eventId],
      executor,
    )
    return row ? eventFromRow(row) : null
  }

  private async findWorkspaceEvent(
    tenantId: string,
    userId: string,
    eventId: string,
    executor: PgExecutor,
  ) {
    const row = await this.maybeOne(
      `SELECT * FROM cloud_workspace_events
       WHERE tenant_id = $1 AND user_id = $2 AND event_id = $3`,
      [tenantId, userId, eventId],
      executor,
    )
    return row ? workspaceEventFromRow(row) : null
  }

  private replayOrRejectEvent(
    existing: SessionEventRecord,
    type: string,
    payload: Record<string, unknown>,
  ) {
    if (existing.type !== type || stableJson(existing.payload) !== stableJson(payload)) {
      throw new Error(`Event id ${existing.eventId} was reused with different content.`)
    }
    return existing
  }

  private replayOrRejectWorkspaceEvent(
    existing: WorkspaceEventRecord,
    type: string,
    payload: Record<string, unknown>,
    expected: {
      sessionId: string | null
      entityType: string
      entityId: string
      operation: string
      projectionVersion: number
    },
  ) {
    if (
      existing.type !== type
      || stableJson(existing.payload) !== stableJson(payload)
      || existing.sessionId !== expected.sessionId
      || existing.entityType !== expected.entityType
      || existing.entityId !== expected.entityId
      || existing.operation !== expected.operation
      || existing.projectionVersion !== expected.projectionVersion
    ) {
      throw new Error(`Workspace event id ${existing.eventId} was reused with different content.`)
    }
    return existing
  }

  private async incrementSessionCounter(
    executor: PgExecutor,
    tenantId: string,
    sessionId: string,
    field: 'next_event_sequence' | 'next_command_sequence',
    updatedAt?: string,
  ) {
    const setUpdatedAt = updatedAt ? ', updated_at = $3' : ''
    const values = updatedAt ? [tenantId, sessionId, updatedAt] : [tenantId, sessionId]
    const result = await executor.query(
      `UPDATE cloud_sessions
       SET ${field} = ${field} + 1${setUpdatedAt}
       WHERE tenant_id = $1 AND session_id = $2
       RETURNING ${field}`,
      values,
    )
    return numberValue(result.rows[0]?.[field])
  }

  private async one<Row extends QueryRow = QueryRow>(
    text: string,
    values?: unknown[],
    executor: PgExecutor = this.pool,
  ) {
    const result = await executor.query<Row>(text, values)
    if (!result.rows[0]) throw new Error('Expected query to return a row.')
    return result.rows[0]
  }

  private async maybeOne<Row extends QueryRow = QueryRow>(
    text: string,
    values?: unknown[],
    executor: PgExecutor = this.pool,
  ) {
    const result = await executor.query<Row>(text, values)
    return result.rows[0] || null
  }

  private async withTransaction<T>(fn: (client: PgClient) => Promise<T>) {
    const client = await this.pool.connect()
    try {
      await client.query('BEGIN')
      const result = await fn(client)
      await client.query('COMMIT')
      return result
    } catch (error) {
      await client.query('ROLLBACK')
      throw error
    } finally {
      client.release()
    }
  }
}

function randomRecordId(prefix: string) {
  return `${prefix}_${Date.now().toString(36)}_${randomBytes(8).toString('base64url')}`
}

export async function createPostgresControlPlaneStore(options: PostgresControlPlaneStoreOptions) {
  return PostgresControlPlaneStore.connect(options)
}
