-
Notifications
You must be signed in to change notification settings - Fork 10
feat(chat/runs): alert-only zombie-owner check on scheduled runs #783
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(); | ||
| }); | ||
| }); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| }); | ||
| }); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| }); | ||
| }); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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<void> { | ||
| 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); | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. P2: A transient Telegram failure suppresses this owner's alert for the full dedup window: the marker is claimed before the send, then retained when Prompt for AI agents |
||
| 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); | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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; | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 <ttl> 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<boolean> { | ||
| 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; | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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<string | null> { | ||
| 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; | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
P2: The
after()callback in the handler firesalertZombieOwner(...)and discards its Promise. When thealertZombieOwnermock rejects (as the second test does), the rejection becomes an unhandled promise rejection in the test environment, which can cause test flakiness or warnings in stricter runners. Add.catch(() => {})to thealertZombieOwnercall insideafter()so both production and test environments are safe regardless of the implementation contract.Prompt for AI agents