From 5117ce8b1524f606afcf0403589dc3b7310da7f3 Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Thu, 23 Jul 2026 06:24:07 -0500 Subject: [PATCH] feat(chat/runs): alert-only zombie-owner check on scheduled runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `handleStartChatRun` starts scheduled runs with no check that a human still uses the account, so an abandoned account keeps generating (and billing) long after the owner left (recoupable/chat#1885). Fire a DEDUPED Telegram alert when the owner's last `role='user'` message is older than 45 days (or they've never sent one). The run is NOT blocked — this is alert-only. - `getLatestUserMessageAt` (`lib/supabase/chat_messages/`) — reads the owner's most recent `role='user'` message via `chat_messages → chats → sessions` inner embeds filtered on `sessions.account_id`. - `isOwnerInactive` — pure 45-day threshold predicate (null → inactive). - `markZombieOwnerAlerted` — per-owner Redis `SET NX EX` dedup marker (30-day window) so daily runs don't spam; fails open on Redis error. - `alertZombieOwner` — orchestrates the three + the Telegram send; never throws. - `handleStartChatRun` — schedules the check via `after()` so it never blocks the 202 or the run. TDD: RED→GREEN for the >45d detection, the dedup marker, the orchestrator, and the handler wiring. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../runs/__tests__/alertZombieOwner.test.ts | 71 +++++++++++++++++++ .../runs/__tests__/handleStartChatRun.test.ts | 26 +++++++ .../runs/__tests__/isOwnerInactive.test.ts | 31 ++++++++ .../__tests__/markZombieOwnerAlerted.test.ts | 38 ++++++++++ lib/chat/runs/alertZombieOwner.ts | 56 +++++++++++++++ lib/chat/runs/handleStartChatRun.ts | 16 ++++- lib/chat/runs/isOwnerInactive.ts | 26 +++++++ lib/chat/runs/markZombieOwnerAlerted.ts | 36 ++++++++++ .../chat_messages/getLatestUserMessageAt.ts | 31 ++++++++ 9 files changed, 330 insertions(+), 1 deletion(-) create mode 100644 lib/chat/runs/__tests__/alertZombieOwner.test.ts create mode 100644 lib/chat/runs/__tests__/isOwnerInactive.test.ts create mode 100644 lib/chat/runs/__tests__/markZombieOwnerAlerted.test.ts create mode 100644 lib/chat/runs/alertZombieOwner.ts create mode 100644 lib/chat/runs/isOwnerInactive.ts create mode 100644 lib/chat/runs/markZombieOwnerAlerted.ts create mode 100644 lib/supabase/chat_messages/getLatestUserMessageAt.ts diff --git a/lib/chat/runs/__tests__/alertZombieOwner.test.ts b/lib/chat/runs/__tests__/alertZombieOwner.test.ts new file mode 100644 index 000000000..b3c824651 --- /dev/null +++ b/lib/chat/runs/__tests__/alertZombieOwner.test.ts @@ -0,0 +1,71 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { alertZombieOwner } from "@/lib/chat/runs/alertZombieOwner"; +import { getLatestUserMessageAt } from "@/lib/supabase/chat_messages/getLatestUserMessageAt"; +import { markZombieOwnerAlerted } from "@/lib/chat/runs/markZombieOwnerAlerted"; +import { sendMessage } from "@/lib/telegram/sendMessage"; + +vi.mock("@/lib/supabase/chat_messages/getLatestUserMessageAt", () => ({ + getLatestUserMessageAt: vi.fn(), +})); +vi.mock("@/lib/chat/runs/markZombieOwnerAlerted", () => ({ + markZombieOwnerAlerted: vi.fn(), +})); +vi.mock("@/lib/telegram/sendMessage", () => ({ + sendMessage: vi.fn(), +})); + +const OLD = new Date(Date.now() - 60 * 24 * 60 * 60 * 1000).toISOString(); +const RECENT = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000).toISOString(); + +describe("alertZombieOwner", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(markZombieOwnerAlerted).mockResolvedValue(true); + vi.mocked(sendMessage).mockResolvedValue({} as never); + }); + + it("alerts when the owner's last user message is older than 45 days", async () => { + vi.mocked(getLatestUserMessageAt).mockResolvedValue(OLD); + + await alertZombieOwner({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }); + + expect(sendMessage).toHaveBeenCalledTimes(1); + const text = vi.mocked(sendMessage).mock.calls[0]?.[0] as string; + expect(text).toContain("acc-1"); + }); + + it("alerts when the owner has never sent a user message", async () => { + vi.mocked(getLatestUserMessageAt).mockResolvedValue(null); + + await alertZombieOwner({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }); + + expect(sendMessage).toHaveBeenCalledTimes(1); + }); + + it("does NOT alert when the owner is still active", async () => { + vi.mocked(getLatestUserMessageAt).mockResolvedValue(RECENT); + + await alertZombieOwner({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }); + + expect(markZombieOwnerAlerted).not.toHaveBeenCalled(); + expect(sendMessage).not.toHaveBeenCalled(); + }); + + it("dedupes — skips the send when the marker was already claimed", async () => { + vi.mocked(getLatestUserMessageAt).mockResolvedValue(OLD); + vi.mocked(markZombieOwnerAlerted).mockResolvedValue(false); + + await alertZombieOwner({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }); + + expect(sendMessage).not.toHaveBeenCalled(); + }); + + it("never throws when a dependency fails (must not break the run)", async () => { + vi.mocked(getLatestUserMessageAt).mockRejectedValue(new Error("db down")); + + await expect( + alertZombieOwner({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }), + ).resolves.toBeUndefined(); + expect(sendMessage).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/chat/runs/__tests__/handleStartChatRun.test.ts b/lib/chat/runs/__tests__/handleStartChatRun.test.ts index 3f1c8555b..9f92a69ac 100644 --- a/lib/chat/runs/__tests__/handleStartChatRun.test.ts +++ b/lib/chat/runs/__tests__/handleStartChatRun.test.ts @@ -8,7 +8,16 @@ import { mintEphemeralAccountKey } from "@/lib/keys/mintEphemeralAccountKey"; import { deleteApiKey } from "@/lib/supabase/account_api_keys/deleteApiKey"; import { buildRunAgentInput } from "@/lib/chat/buildRunAgentInput"; import { start } from "workflow/api"; +import { alertZombieOwner } from "@/lib/chat/runs/alertZombieOwner"; +// `after()` schedules post-response work; run the callback synchronously in tests. +vi.mock("next/server", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, after: (fn: () => unknown) => fn() }; +}); +vi.mock("@/lib/chat/runs/alertZombieOwner", () => ({ + alertZombieOwner: vi.fn(), +})); vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({ "Access-Control-Allow-Origin": "*" })), })); @@ -98,6 +107,23 @@ describe("handleStartChatRun", () => { expect(deleteApiKey).not.toHaveBeenCalled(); }); + it("fires the zombie-owner alert (alert-only) for the run's owner after starting", async () => { + const res = await handleStartChatRun(req()); + + expect(res.status).toBe(202); + expect(alertZombieOwner).toHaveBeenCalledWith( + expect.objectContaining({ accountId: "acc-1", chatId: "chat-1", sessionId: "sess-1" }), + ); + }); + + it("does not block the run when the zombie-owner alert throws", async () => { + vi.mocked(alertZombieOwner).mockRejectedValueOnce(new Error("alert boom")); + + const res = await handleStartChatRun(req()); + + expect(res.status).toBe(202); + }); + it("returns the validation error short-circuit", async () => { vi.mocked(validateChatRunRequest).mockResolvedValue( NextResponse.json({ status: "error" }, { status: 401 }), diff --git a/lib/chat/runs/__tests__/isOwnerInactive.test.ts b/lib/chat/runs/__tests__/isOwnerInactive.test.ts new file mode 100644 index 000000000..429c42fa2 --- /dev/null +++ b/lib/chat/runs/__tests__/isOwnerInactive.test.ts @@ -0,0 +1,31 @@ +import { describe, it, expect } from "vitest"; +import { isOwnerInactive, ZOMBIE_OWNER_INACTIVE_DAYS } from "@/lib/chat/runs/isOwnerInactive"; + +const now = new Date("2026-07-23T00:00:00.000Z"); +const daysAgo = (n: number) => new Date(now.getTime() - n * 24 * 60 * 60 * 1000).toISOString(); + +describe("isOwnerInactive", () => { + it("treats a missing last-message timestamp as inactive (no human message ever)", () => { + expect(isOwnerInactive(null, now)).toBe(true); + }); + + it("is inactive when the last user message is older than the threshold", () => { + expect(isOwnerInactive(daysAgo(ZOMBIE_OWNER_INACTIVE_DAYS + 1), now)).toBe(true); + }); + + it("is active when the last user message is within the threshold", () => { + expect(isOwnerInactive(daysAgo(ZOMBIE_OWNER_INACTIVE_DAYS - 1), now)).toBe(false); + }); + + it("is active for a message exactly at the threshold (not yet stale)", () => { + expect(isOwnerInactive(daysAgo(ZOMBIE_OWNER_INACTIVE_DAYS), now)).toBe(false); + }); + + it("is active for a message from today", () => { + expect(isOwnerInactive(now.toISOString(), now)).toBe(false); + }); + + it("uses a 45-day threshold", () => { + expect(ZOMBIE_OWNER_INACTIVE_DAYS).toBe(45); + }); +}); diff --git a/lib/chat/runs/__tests__/markZombieOwnerAlerted.test.ts b/lib/chat/runs/__tests__/markZombieOwnerAlerted.test.ts new file mode 100644 index 000000000..5bd0ce091 --- /dev/null +++ b/lib/chat/runs/__tests__/markZombieOwnerAlerted.test.ts @@ -0,0 +1,38 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { markZombieOwnerAlerted } from "@/lib/chat/runs/markZombieOwnerAlerted"; +import redis from "@/lib/redis/connection"; + +vi.mock("@/lib/redis/connection", () => ({ + default: { set: vi.fn() }, +})); + +describe("markZombieOwnerAlerted", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("claims the marker with SET NX + a TTL and returns true on first alert", async () => { + vi.mocked(redis.set).mockResolvedValue("OK"); + + const shouldSend = await markZombieOwnerAlerted("acc-1"); + + expect(shouldSend).toBe(true); + const args = vi.mocked(redis.set).mock.calls[0]; + expect(args[0]).toContain("acc-1"); + // NX so a concurrent/repeat run can't re-claim; EX so it auto-expires. + expect(args).toContain("NX"); + expect(args).toContain("EX"); + }); + + it("returns false when the marker already exists (deduped — skip the alert)", async () => { + vi.mocked(redis.set).mockResolvedValue(null); + + expect(await markZombieOwnerAlerted("acc-1")).toBe(false); + }); + + it("returns true (fails open) when Redis errors — never silences a real alert on infra blips", async () => { + vi.mocked(redis.set).mockRejectedValue(new Error("redis down")); + + expect(await markZombieOwnerAlerted("acc-1")).toBe(true); + }); +}); diff --git a/lib/chat/runs/alertZombieOwner.ts b/lib/chat/runs/alertZombieOwner.ts new file mode 100644 index 000000000..8fb040395 --- /dev/null +++ b/lib/chat/runs/alertZombieOwner.ts @@ -0,0 +1,56 @@ +import { getLatestUserMessageAt } from "@/lib/supabase/chat_messages/getLatestUserMessageAt"; +import { isOwnerInactive, ZOMBIE_OWNER_INACTIVE_DAYS } from "@/lib/chat/runs/isOwnerInactive"; +import { markZombieOwnerAlerted } from "@/lib/chat/runs/markZombieOwnerAlerted"; +import { sendMessage } from "@/lib/telegram/sendMessage"; + +export type ZombieOwnerAlertParams = { + /** Account whose scheduled run just started. */ + accountId: string; + /** Chat the run belongs to. */ + chatId: string; + /** Session the run belongs to. */ + sessionId: string; +}; + +/** + * Alert-only zombie-owner check for scheduled runs (recoupable/chat#1885). + * + * `handleStartChatRun` starts scheduled runs with no check that a human still + * uses the account. This fires a DEDUPED Telegram alert when the owner's last + * `role='user'` message is older than {@link ZOMBIE_OWNER_INACTIVE_DAYS} days + * (or they've never sent one). The run is NOT blocked — this only surfaces + * accounts that keep generating long after the human left. + * + * Dedup: a per-owner Redis marker (`markZombieOwnerAlerted`) so daily runs for + * the same dormant account alert at most once per window. Never throws — + * schedule it via `after()` so it can't break the run or delay the response. + */ +export async function alertZombieOwner(params: ZombieOwnerAlertParams): Promise { + const { accountId, chatId, sessionId } = params; + + try { + const lastUserMessageAt = await getLatestUserMessageAt(accountId); + if (!isOwnerInactive(lastUserMessageAt, new Date())) return; + + // Dedup before sending so repeated scheduled runs don't spam. + const shouldSend = await markZombieOwnerAlerted(accountId); + if (!shouldSend) return; + + const lastSeen = lastUserMessageAt ?? "never"; + const text = [ + "🧟 *Zombie-owner scheduled run*", + "", + `*Account:* ${accountId}`, + `*Chat:* ${chatId}`, + `*Session:* ${sessionId}`, + `*Last user message:* ${lastSeen}`, + "", + `No human \`role='user'\` message in > ${ZOMBIE_OWNER_INACTIVE_DAYS} days, but scheduled runs are still firing.`, + `*Time:* ${new Date().toISOString()}`, + ].join("\n"); + + await sendMessage(text, { parse_mode: "Markdown" }); + } catch (error) { + console.error("[alertZombieOwner] failed (non-blocking):", error); + } +} diff --git a/lib/chat/runs/handleStartChatRun.ts b/lib/chat/runs/handleStartChatRun.ts index 41b68d1c7..2eba2aa7b 100644 --- a/lib/chat/runs/handleStartChatRun.ts +++ b/lib/chat/runs/handleStartChatRun.ts @@ -1,6 +1,7 @@ -import { NextRequest, NextResponse } from "next/server"; +import { NextRequest, NextResponse, after } from "next/server"; import { start } from "workflow/api"; import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; +import { alertZombieOwner } from "@/lib/chat/runs/alertZombieOwner"; import { errorResponse } from "@/lib/networking/errorResponse"; import { validateChatRunRequest } from "@/lib/chat/runs/validateChatRunRequest"; import { provisionRunSession } from "@/lib/chat/runs/provisionRunSession"; @@ -69,6 +70,19 @@ export async function handleStartChatRun(request: NextRequest): Promise 45 days old. Runs + // via `after()` so it never blocks the 202 or the run itself, and swallows + // its own errors. + after(() => + alertZombieOwner({ + accountId, + chatId: provisioned.chat.id, + sessionId: provisioned.session.id, + }), + ); + // Return the run handle plus the persisted-output identifiers so the caller // can read the result later (the workflow runId alone can't be resolved back // to the chat): GET /api/chat/{chatId}/stream resumes the stream, and the diff --git a/lib/chat/runs/isOwnerInactive.ts b/lib/chat/runs/isOwnerInactive.ts new file mode 100644 index 000000000..7ccf11fb6 --- /dev/null +++ b/lib/chat/runs/isOwnerInactive.ts @@ -0,0 +1,26 @@ +/** + * Age (in days) past which an account's owner is considered inactive — no + * `role='user'` message in this long means a human likely stopped using the + * account, yet its scheduled runs keep firing (recoupable/chat#1885). + */ +export const ZOMBIE_OWNER_INACTIVE_DAYS = 45; + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** + * Pure predicate: is the account owner inactive as of `now`? + * + * `null` (the owner has never sent a user message) counts as inactive. A + * message exactly at the threshold is still considered active — only strictly + * older than {@link ZOMBIE_OWNER_INACTIVE_DAYS} is stale. + * + * @param lastUserMessageAt - ISO timestamp of the owner's most recent + * `role='user'` message, or `null` if none exists. + * @param now - Reference time (injected for deterministic tests). + */ +export function isOwnerInactive(lastUserMessageAt: string | null, now: Date): boolean { + if (!lastUserMessageAt) return true; + + const ageMs = now.getTime() - new Date(lastUserMessageAt).getTime(); + return ageMs > ZOMBIE_OWNER_INACTIVE_DAYS * DAY_MS; +} diff --git a/lib/chat/runs/markZombieOwnerAlerted.ts b/lib/chat/runs/markZombieOwnerAlerted.ts new file mode 100644 index 000000000..45246e1f5 --- /dev/null +++ b/lib/chat/runs/markZombieOwnerAlerted.ts @@ -0,0 +1,36 @@ +import redis from "@/lib/redis/connection"; + +/** + * How long a zombie-owner alert stays deduped per owner. Scheduled runs can + * fire daily; without a window every run would re-alert the same dormant + * account. 30 days means at most one alert per owner per month. + */ +const ZOMBIE_OWNER_ALERT_DEDUP_SECONDS = 30 * 24 * 60 * 60; + +const markerKey = (accountId: string) => `zombie-owner-alert:${accountId}`; + +/** + * Atomically claim the per-owner alert marker in Redis. Returns `true` when + * this caller won the claim (first alert in the dedup window → send it) and + * `false` when the marker already existed (a recent run already alerted → skip). + * + * `SET key 1 EX NX` is atomic, so concurrent scheduled runs for the same + * owner can't both send. Fails OPEN (returns `true`) on a Redis error so an + * infra blip never silences a genuine alert — a duplicate alert is cheaper than + * a missed one. + */ +export async function markZombieOwnerAlerted(accountId: string): Promise { + try { + const result = await redis.set( + markerKey(accountId), + "1", + "EX", + ZOMBIE_OWNER_ALERT_DEDUP_SECONDS, + "NX", + ); + return result === "OK"; + } catch (error) { + console.error("[markZombieOwnerAlerted] redis error, failing open:", error); + return true; + } +} diff --git a/lib/supabase/chat_messages/getLatestUserMessageAt.ts b/lib/supabase/chat_messages/getLatestUserMessageAt.ts new file mode 100644 index 000000000..e77a6bb8d --- /dev/null +++ b/lib/supabase/chat_messages/getLatestUserMessageAt.ts @@ -0,0 +1,31 @@ +import supabase from "@/lib/supabase/serverClient"; + +/** + * Return the ISO `created_at` of an account owner's most recent `role='user'` + * chat message, or `null` if they've never sent one (or on DB error — callers + * treat "unknown" as "no recent human activity", which is the safe default for + * an alert-only signal). + * + * Walks `chat_messages → chats → sessions` via PostgREST inner embeds and + * filters on `sessions.account_id`, so it counts only messages the owner + * actually authored across their own sessions. Used by the zombie-owner alert + * (recoupable/chat#1885) to detect accounts whose scheduled runs keep firing + * long after the human stopped engaging. + */ +export async function getLatestUserMessageAt(accountId: string): Promise { + const { data, error } = await supabase + .from("chat_messages") + .select("created_at, chats!inner(sessions!inner(account_id))") + .eq("role", "user") + .eq("chats.sessions.account_id", accountId) + .order("created_at", { ascending: false }) + .limit(1) + .maybeSingle(); + + if (error) { + console.error("[getLatestUserMessageAt] error:", error); + return null; + } + + return data?.created_at ?? null; +}