Ask the quota question where the writing happens

Three of the five checks in direct_messages_validate were send quotas:
a five second gap, twenty an hour, a hundred a day. They are not
answers about a row, they are answers about a moment — whether this
agent may send one more right now — and a trigger answers them for
every insert, including a restore replaying years of history.

The restore already suspends this gate, so nothing was broken. But not
having to suspend it is better than suspending it, and the follow
limits were never triggers for exactly this reason. The quotas now sit
next to them on the write path, checked after idempotent replay so a
repeated request is still the same message rather than a new one.

What this costs: two genuinely simultaneous requests from one agent can
both pass the count. One extra message is not a hole — the quota exists
to stop a flood and still does.

The trigger keeps the two availability checks. They are cheap and they
back up the API's own resolution.

The burst limit had a test, but one that passed while the quota was
still in the trigger, so it proved nothing about where the rule lives.
The hourly and daily limits had no test at all. All three now do, on
their own agents and their own time window, and they check that a
rejected send writes nothing and does not burn its idempotency key.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: sametbasbug

migrations/0027_direct_message_quota_moves_to_the_write_path.sql

@@ -0,0 +1,39 @@
+-- DM gönderim kotaları tetikleyiciden yazma yoluna taşınıyor.
+--
+-- direct_messages_validate beş şey birden denetliyordu: gönderen uygun mu,
+-- alıcı uygun mu, ve üç kota — beş saniyelik ara, saatlik 20, günlük 100.
+-- İlk ikisi bir satırın anlamlı olup olmadığını soruyor. Son üçü ise "bu ajan
+-- şu an bir mesaj daha gönderebilir mi" diye soruyor ki bu satırın değil
+-- eylemin sorusu, ve cevabı yalnız yazma anında geçerli.
+--
+-- Kotanın burada durmasının somut bedeli yedeklerdi: tetikleyici geri yükleme
+-- sırasında da çalışıyor ve geçmiş kendi limitine takılıyor. Geri yükleme
+-- artık bu kapıyı askıya alıyor, ama kapıyı askıya almak zorunda kalmamak
+-- daha iyi — kota zaten oraya ait değildi. Takip limitleri de aynı sebeple
+-- hiç tetikleyici olmadı.
+--
+-- Kalan iki denetim yerinde: bunlar ucuz ve API'nin kendi kontrolünü
+-- yedekliyorlar. Geri yüklemede hâlâ askıya alınıyorlar, çünkü "şu an aktif
+-- mi" de bugüne ait bir soru.
+
+DROP TRIGGER direct_messages_validate;
+
+CREATE TRIGGER direct_messages_validate
+BEFORE INSERT ON direct_messages
+BEGIN
+  SELECT RAISE(ABORT, 'direct_message_sender_unavailable')
+  WHERE NOT EXISTS (
+    SELECT 1 FROM agents
+    WHERE id = NEW.sender_agent_id
+      AND status = 'active'
+      AND onboarding_state = 'active'
+  );
+
+  SELECT RAISE(ABORT, 'direct_message_recipient_unavailable')
+  WHERE NOT EXISTS (
+    SELECT 1 FROM agents
+    WHERE id = NEW.recipient_agent_id
+      AND status = 'active'
+      AND onboarding_state = 'active'
+  );
+END;

scripts/orbit-slice5-tests.ts

@@ -689,6 +689,90 @@ describe('Orbit V6 Slice 5 dashboard and platform core', { concurrency: false },
     });
   });
 
+  /* Kotalar tetikleyicide değil yazma yolunda. Burst'ü yukarıdaki test zaten
+   * görüyordu — ama o test kota tetikleyicideyken de geçiyordu, yani taşımayı
+   * kanıtlamıyor. Saatlik ve günlük sınırlar ise hiç sınanmamıştı; artık
+   * benim yazdığım kodda yaşıyorlar. */
+  test('direct message quotas are enforced on the write path and write nothing when they reject', async () => {
+    // Kendi ajanları: bu test yüzden fazla mesaj yazıyor ve paylaşılan bir
+    // gönderenin kutusunu kullansa sonraki testlerin aradığı mesajı ilk
+    // sayfadan iterdi.
+    const sender = await seedAgent('slice5-dm-quota-sender');
+    const recipient = await seedAgent('slice5-dm-quota-recipient');
+    // Kendi penceresi: başka testlerin mesajları ne saatlik ne günlük sayıma
+    // karışsın.
+    const base = NOW + 48 * 60 * 60 * 1000;
+    const hour = 60 * 60 * 1000;
+
+    const send = async (at: number, key: string) => await fetch(`${baseUrl}/v1/direct-messages`, {
+      method: 'POST',
+      headers: {
+        authorization: `Bearer ${sender.token}`,
+        'content-type': 'application/json',
+        'x-test-now': String(at),
+        'idempotency-key': key,
+      },
+      body: JSON.stringify({
+        recipientHandle: recipient.handle,
+        bodyMarkdown: `Kota penceresi mesajı ${key}.`,
+      }),
+    });
+    const sentCount = async () => {
+      const response = await fetch(`${baseUrl}/v1/direct-messages?box=sent&limit=50`, {
+        headers: { authorization: `Bearer ${sender.token}`, 'x-test-now': String(base) },
+      });
+      return (await response.json() as { directMessages: unknown[] }).directMessages.length;
+    };
+
+    // Beş saniyelik ara: reddedilen istek hiçbir şey yazmıyor ve kendi
+    // idempotency anahtarını da tüketmiyor.
+    assert.equal((await send(base, 'dm-window-1')).status, 201);
+    const before = await sentCount();
+    const tooSoon = await send(base + 4_999, 'dm-window-2');
+    assert.equal(tooSoon.status, 429);
+    assert.equal((await tooSoon.json() as { error: { code: string } }).error.code, 'direct_message_burst_limited');
+    assert.equal(await sentCount(), before);
+    assert.equal((await send(base + 5_000, 'dm-window-2')).status, 201);
+
+    // Saatlik sınır. İki mesaj gitti, on sekiz tane daha yirmiye tamamlıyor.
+    for (let index = 0; index < 18; index += 1) {
+      const at = base + 10_000 + index * 6_000;
+      assert.equal((await send(at, `dm-hour-${index}`)).status, 201, `saatlik dolum ${index}`);
+    }
+    const hourly = await send(base + 10_000 + 18 * 6_000, 'dm-hour-over');
+    assert.equal(hourly.status, 429);
+    const hourlyError = (await hourly.json() as { error: { code: string; details: { quota: unknown } } }).error;
+    assert.equal(hourlyError.code, 'direct_message_hourly_limit_exceeded');
+    assert.deepEqual(hourlyError.details.quota, {
+      key: 'direct_message.send.rolling_hour',
+      limit: 20,
+      remaining: 0,
+      windowSeconds: 3_600,
+      resetAt: base + hour,
+    });
+
+    // Günlük sınır. İki saat aralıklı bloklar: her blok başında bir önceki
+    // blok saatlik pencerenin dışında kalıyor, yani yalnız günlük sayım
+    // birikiyor. Dört blok daha yüze tamamlıyor.
+    for (let block = 1; block <= 4; block += 1) {
+      for (let index = 0; index < 20; index += 1) {
+        const at = base + block * 2 * hour + index * 6_000;
+        assert.equal((await send(at, `dm-day-${block}-${index}`)).status, 201, `günlük dolum ${block}/${index}`);
+      }
+    }
+    const daily = await send(base + 10 * hour, 'dm-day-over');
+    assert.equal(daily.status, 429);
+    const dailyError = (await daily.json() as { error: { code: string; details: { quota: unknown } } }).error;
+    assert.equal(dailyError.code, 'direct_message_daily_limit_exceeded');
+    assert.deepEqual(dailyError.details.quota, {
+      key: 'direct_message.send.rolling_day',
+      limit: 100,
+      remaining: 0,
+      windowSeconds: 86_400,
+      resetAt: base + 24 * hour,
+    });
+  });
+
   test('sponsor witnesses only their own agent and reading changes nothing', async () => {
     const sender = agents.get('slice5-equinox')!;
     const recipient = agents.get('slice5-external')!;

src/server/http/api.ts

@@ -2461,61 +2461,79 @@ async function markDirectMessageReadForAgent(
   return json({ directMessage: { id: messageId, readAt } });
 }
 
-async function directMessageQuotaApiError(
+/* DM gönderim kotaları.
+ *
+ * Bunlar bir zamanlar direct_messages üzerinde tetikleyiciydi ve orada
+ * durmalarının bedelini yedeklerde ödedik: tetikleyici geri yüklemede de
+ * çalışıyor, yani geçmiş kendi limitine takılıyordu. Ama asıl sorun teknik
+ * değil, kavramsal — kota "bu satır geçerli mi" sorusunun cevabı değil, "bu
+ * ajan şu an bir mesaj daha gönderebilir mi" sorusunun cevabı. O soru yazma
+ * anına ait ve yazma yolunda sorulmalı. Takip limitleri de aynı yerde duruyor.
+ *
+ * Bunun kabul ettiğimiz bedeli şu: aynı ajandan tam eşzamanlı iki istek
+ * ikisi de sayımı geçebilir. Fazladan bir mesaj sızması bir güvenlik açığı
+ * değil, kotanın amacı sel basmasını önlemek ve onu önlüyor. Karşılığında
+ * yedek geri yüklenebiliyor.
+ */
+const DIRECT_MESSAGE_MIN_INTERVAL_MS = 5_000;
+const DIRECT_MESSAGE_LIMIT_PER_HOUR = 20;
+const DIRECT_MESSAGE_LIMIT_PER_DAY = 100;
+
+async function requireDirectMessageQuota(
   repository: DirectMessageRepository,
   agentId: string,
   now: number,
-  error: unknown,
-): Promise<ApiError | null> {
-  const message = error instanceof Error ? error.message : String(error);
-  if (!/direct_message_(?:burst|hourly|daily)_limit_exceeded/u.test(message)) {
-    return null;
-  }
+): Promise<void> {
   const state = await repository.getSendRecoveryState(agentId, now);
-  if (message.includes('burst')) {
-    return quotaError(
+  if (
+    state.lastMessageAt !== null
+    && now - state.lastMessageAt < DIRECT_MESSAGE_MIN_INTERVAL_MS
+  ) {
+    throw quotaError(
       now,
       'direct_message_burst_limited',
       'The agent reached a direct-message rate limit.',
       {
         key: 'direct_message.send.minimum_interval',
         limit: 1,
         remaining: 0,
-        windowSeconds: 5,
-        resetAt: Math.max(now + 1, (state.lastMessageAt ?? now) + 5000),
+        windowSeconds: DIRECT_MESSAGE_MIN_INTERVAL_MS / 1000,
+        resetAt: Math.max(now + 1, state.lastMessageAt + DIRECT_MESSAGE_MIN_INTERVAL_MS),
       },
     );
   }
-  if (message.includes('hourly')) {
-    return quotaError(
+  if (state.hourlyCount >= DIRECT_MESSAGE_LIMIT_PER_HOUR) {
+    throw quotaError(
       now,
       'direct_message_hourly_limit_exceeded',
       'The agent reached a direct-message rate limit.',
       {
         key: 'direct_message.send.rolling_hour',
-        limit: 20,
-        remaining: Math.max(0, 20 - state.hourlyCount),
+        limit: DIRECT_MESSAGE_LIMIT_PER_HOUR,
+        remaining: Math.max(0, DIRECT_MESSAGE_LIMIT_PER_HOUR - state.hourlyCount),
         windowSeconds: 60 * 60,
         resetAt: state.oldestHourlyMessageAt === null
           ? now + 60 * 60 * 1000
           : state.oldestHourlyMessageAt + 60 * 60 * 1000,
       },
     );
   }
-  return quotaError(
-    now,
-    'direct_message_daily_limit_exceeded',
-    'The agent reached a direct-message rate limit.',
-    {
-      key: 'direct_message.send.rolling_day',
-      limit: 100,
-      remaining: Math.max(0, 100 - state.dailyCount),
-      windowSeconds: 24 * 60 * 60,
-      resetAt: state.oldestDailyMessageAt === null
-        ? now + 24 * 60 * 60 * 1000
-        : state.oldestDailyMessageAt + 24 * 60 * 60 * 1000,
-    },
-  );
+  if (state.dailyCount >= DIRECT_MESSAGE_LIMIT_PER_DAY) {
+    throw quotaError(
+      now,
+      'direct_message_daily_limit_exceeded',
+      'The agent reached a direct-message rate limit.',
+      {
+        key: 'direct_message.send.rolling_day',
+        limit: DIRECT_MESSAGE_LIMIT_PER_DAY,
+        remaining: Math.max(0, DIRECT_MESSAGE_LIMIT_PER_DAY - state.dailyCount),
+        windowSeconds: 24 * 60 * 60,
+        resetAt: state.oldestDailyMessageAt === null
+          ? now + 24 * 60 * 60 * 1000
+          : state.oldestDailyMessageAt + 24 * 60 * 60 * 1000,
+      },
+    );
+  }
 }
 
 async function handleSendDirectMessageForPrincipal(
@@ -2550,6 +2568,10 @@ async function handleSendDirectMessageForPrincipal(
   );
   if (idem.replay) return replayResponse(idem.replay);
   await requireCriticalAnnouncementsRead(platformRepository, principal, now);
+  // Kota tekrar oynatmadan sonra bakılıyor: aynı isteğin tekrarı yeni bir
+  // mesaj değil, zaten gönderilmiş olanın tekrar söylenmesi. Önce baksaydık
+  // her başarılı gönderimin kendi tekrarını beş saniye boyunca reddederdik.
+  await requireDirectMessageQuota(directMessageRepository, principal.agentId, now);
 
   const message: DirectMessageView = {
     id: createEntityId(),
@@ -2562,37 +2584,25 @@ async function handleSendDirectMessageForPrincipal(
     readAt: null,
   };
   const responseBody = { directMessage: directMessageResponse(message) };
-  let concurrentReplay: Response | null;
-  try {
-    concurrentReplay = await runIdempotentMutation(
-      publicationRepository,
-      'agent',
-      principal.agentId,
-      idem.keyDigest,
-      idem.requestDigest,
-      () => directMessageRepository.sendMessage({
-        message,
-        idempotency: {
-          ...idem.row,
-          principalType: 'agent',
-          principalId: principal.agentId,
-          responseStatus: 201,
-          responseJson: canonicalJson(responseBody),
-        },
-        auditEventId: createEntityId(),
-        requestId,
-      }),
-    );
-  } catch (error) {
-    const mapped = await directMessageQuotaApiError(
-      directMessageRepository,
-      principal.agentId,
-      now,
-      error,
-    );
-    if (mapped) throw mapped;
-    throw error;
-  }
+  const concurrentReplay = await runIdempotentMutation(
+    publicationRepository,
+    'agent',
+    principal.agentId,
+    idem.keyDigest,
+    idem.requestDigest,
+    () => directMessageRepository.sendMessage({
+      message,
+      idempotency: {
+        ...idem.row,
+        principalType: 'agent',
+        principalId: principal.agentId,
+        responseStatus: 201,
+        responseJson: canonicalJson(responseBody),
+      },
+      auditEventId: createEntityId(),
+      requestId,
+    }),
+  );
   if (concurrentReplay) return concurrentReplay;
   return idempotentJson(responseBody, 201, idem.row.expiresAt);
 }