diff --git a/.env.example b/.env.example index 74c111e..cf777a8 100644 --- a/.env.example +++ b/.env.example @@ -68,3 +68,9 @@ LINK_TOKEN_ENCRYPTION_KEY= # Exact trusted checkout/frame origins per approved merchant (JSON object). LINK_CHECKOUT_ORIGINS={} + +# Optional persistent public-web watches. Apply migrations/004_parallel_monitors.sql and 005_monitor_frequency.sql. +# Copy the Parallel organization's secret from Settings > Webhooks. +PARALLEL_WEBHOOK_SECRET= +# Generate a random worker bearer secret; configure a scheduler separately. +CRON_SECRET= diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 486903a..3a1169d 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -31,6 +31,7 @@ jobs: --health-timeout 5s --health-retries 10 env: + TEST_MONITOR_DATABASE_URL: postgresql://bell_test:bell_test@127.0.0.1:5432/bell_test TEST_LINK_DATABASE_URL: postgresql://bell_test:bell_test@127.0.0.1:5432/bell_test steps: - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 diff --git a/README.md b/README.md index d096233..56433a1 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ pnpm dev | --- | --- | --- | | Browserbase + Stagehand | Browser sessions, automation, saved Contexts, and human handoff | [Browser runtime](docs/browser-runtime.md) | | Eve by Vercel | Agent configuration, tools, conversation state, and channels | [Eve in Bell](docs/repository.md#eve-by-vercel) | -| Parallel | Public-web search, source excerpts, and citations | Set `PARALLEL_API_KEY` in `.env` | +| Parallel | Public-web research and persistent watches with configurable frequency | Set `PARALLEL_API_KEY` in `.env`; [watch setup](docs/monitors.md) | | Linq | iMessage and SMS through a signed webhook and sender allowlist | [Messaging setup](docs/linq.md) | | Link by Stripe | Wallet authorization, spend approval, and checkout through Linq | [Sandbox and wallet setup](docs/link.md) | | 1Password | Operator credentials and OTP; personal Agentic Autofill is disabled by default | [Operator access](docs/browser-runtime.md#credentials-and-otp) · [Personal access status](docs/onepassword.md) | diff --git a/agent/channels/linq.ts b/agent/channels/linq.ts index fe3a44c..db40b4c 100644 --- a/agent/channels/linq.ts +++ b/agent/channels/linq.ts @@ -7,6 +7,7 @@ const inboundMessage = z.object({ direction: z.literal("inbound"), chat: z.object({ is_group: z.literal(false), + id: z.string().min(1).max(256), owner_handle: z.object({ handle: phoneNumber }), }), sender_handle: z.object({ handle: phoneNumber, is_me: z.literal(false) }), @@ -37,8 +38,10 @@ export default linqChannel({ turnPolicy: "steer", onMessage(_context, message) { if (!isAllowedBellMessage(message)) return null; + const auth = defaultLinqAuth(message); + const incoming = inboundMessage.parse(message.raw); return { - auth: defaultLinqAuth(message), + auth: { ...auth, attributes: { ...auth.attributes, bell_watch_chat: incoming.chat.id, bell_watch_sender: incoming.sender_handle.handle, bell_watch_line: incoming.chat.owner_handle.handle } }, context: [ "This request arrived through Bell's private Linq iMessage or SMS channel. " + "Reply concisely in this chat. Use the same browser for follow-up requests. " diff --git a/agent/instructions.md b/agent/instructions.md index 389d537..7d6734a 100644 --- a/agent/instructions.md +++ b/agent/instructions.md @@ -47,3 +47,18 @@ navigation needs a fresh proof. After access, call snapshot and screenshot to ve Do not claim a successful mint or merchant access unless the tool result confirms it. For private Linq senders, use `onepassword_connect`, `onepassword_status`, `onepassword_login`, and `onepassword_disconnect` for personal 1Password access. If status is disabled, explain that personal 1Password login is not available yet; do not offer a consent link or fall back to the operator vault. Never use the operator vault tools for a messaging sender. Connecting is separate from approving a website login. Use the exact login URL and a distinct expected success path on the same origin. A verified destination is not proof of account identity. The first version supports single-step sign-in only. Approval links are sent through Eve authorization; do not ask for tokens, integration keys, passwords, or OTPs in chat. If a private fill fails, start a new conversation rather than inspecting the blocked browser. Personal browser mode does not support reusable Contexts. + +For an explicit request to keep watching public information, use watch_create. Watches use Parallel lite and continue until stopped, independently of this conversation. +Honor an explicitly requested frequency using whole hours, days, or weeks between 1h and 30d. +When no frequency is specified, default to daily and say so. Never silently round an unsupported +request: explain the limits and ask the user to choose a supported interval. Confirm the selected +frequency and explain that faster checks increase scheduled usage. Research frequency is not an +alert guarantee; send updates only for detected changes. Use watch_update to change an existing +watch's frequency without replacing it. A non-null pending_frequency is a queued or unresolved +change, not confirmation; use watch_list to check completion and failures. Never include secrets or +private browser content in a watch query. Use watch_list for recent updates and recovery status. +watch_list returns one page of watches: when its next_cursor is not null, pass that back as cursor +before concluding that a watch does not exist, or narrow the search with status. +If creation is uncertain, list the existing watch instead of creating a replacement. A pending +watch is not confirmed active. Use watch_stop to cancel permanently; cancel_pending means +alerts are suppressed locally but upstream execution and billing may continue until confirmed. diff --git a/agent/lib/monitors.ts b/agent/lib/monitors.ts new file mode 100644 index 0000000..6f785ba --- /dev/null +++ b/agent/lib/monitors.ts @@ -0,0 +1,23 @@ +import type { ToolContext } from "eve/tools"; +import { assertOwner, destinationSchema } from "../../server/monitors/config"; + +export function watchIdentity(ctx: ToolContext) { + const auth = ctx.session.auth.current; + if (!auth || auth.authenticator !== "linq-message" || auth.principalType !== "user") throw new Error("Watches require an authenticated Linq sender."); + const owner = { issuer: auth.issuer ?? "", id: auth.principalId }; + assertOwner(owner); + return owner; +} +export function watchDestination(ctx: ToolContext) { + watchIdentity(ctx); + const attributes = ctx.session.auth.current!.attributes; + return destinationSchema.parse({ chat: attributes.bell_watch_chat, sender: attributes.bell_watch_sender, line: attributes.bell_watch_line }); +} +export async function safeWatch(operation: () => Promise): Promise { + try { return await operation(); } + catch (error) { + const message = error instanceof Error ? error.message : ""; + if (/^(Watches |Watch not found|Watch operation|This Linq)/.test(message)) throw new Error(message); + throw new Error("Watch operation failed. List watches before retrying creation; an uncertain watch may already exist."); + } +} diff --git a/agent/tools/watch_create.ts b/agent/tools/watch_create.ts new file mode 100644 index 0000000..9178634 --- /dev/null +++ b/agent/tools/watch_create.ts @@ -0,0 +1,16 @@ +import { defineTool, toolOutput } from "eve/tools"; +import { z } from "zod"; +import { monitors } from "../../server/monitors/service"; +import { watchIdentity, safeWatch, watchDestination } from "../lib/monitors"; +import { queryInput, frequencyInput } from "../../server/monitors/config"; + +export default defineTool({ + description: "Create a public-web watch with Parallel lite. Frequency defaults to daily; whole hours/days/weeks from 1h to 30d are supported. Faster checks increase scheduled usage. Only create when the user asks for ongoing monitoring. Never include secrets or private browser content. Watches continue until stopped; list existing watches before retrying an uncertain creation.", + inputSchema: z.object({ query: queryInput, frequency: frequencyInput.default("1d") }), + execute: (input, ctx) => safeWatch(() => { + watchIdentity(ctx); + const service = monitors(); + return service.create(watchIdentity(ctx), watchDestination(ctx), JSON.stringify([ctx.session.id, ctx.callId]), input.query, input.frequency); + }), + toModelOutput: output => toolOutput.json(output), +}); diff --git a/agent/tools/watch_list.ts b/agent/tools/watch_list.ts new file mode 100644 index 0000000..573271d --- /dev/null +++ b/agent/tools/watch_list.ts @@ -0,0 +1,14 @@ +import { defineTool, toolOutput } from "eve/tools"; +import { z } from "zod"; +import { monitors } from "../../server/monitors/service"; +import { watchIdentity, safeWatch } from "../lib/monitors"; + +export default defineTool({ + description: "List this sender's watches, recent updates, and pending failures. Use this to reconcile uncertain creation before making another watch, or to find the ID of a watch to update or stop. Returns one page of 50 watches, open ones (still running or still needing repair) before cancelled and failed ones, newest first within each. When next_cursor is not null there are older watches: call again with that value as cursor to read the next page.", + inputSchema: z.object({ + status: z.enum(["all", "open", "closed"]).default("all").describe("Which watches to return: all, only open ones (not cancelled or failed), or only closed ones."), + cursor: z.string().max(512).optional().describe("The next_cursor from a previous watch_list call, to read the next page of older watches."), + }), + execute: (input, ctx) => safeWatch(() => monitors().list(watchIdentity(ctx), { status: input.status, cursor: input.cursor })), + toModelOutput: output => toolOutput.json(output), +}); diff --git a/agent/tools/watch_stop.ts b/agent/tools/watch_stop.ts new file mode 100644 index 0000000..b282e1a --- /dev/null +++ b/agent/tools/watch_stop.ts @@ -0,0 +1,15 @@ +import { defineTool, toolOutput } from "eve/tools"; +import { z } from "zod"; +import { monitors } from "../../server/monitors/service"; +import { watchIdentity, safeWatch } from "../lib/monitors"; + +export default defineTool({ + description: "Permanently stop one of this sender's watches. Local alerts stop immediately; cancel_pending means upstream cancellation is still pending and billing may continue.", + inputSchema: z.object({ id: z.string().uuid() }), + execute: (input, ctx) => safeWatch(() => { + watchIdentity(ctx); + const service = monitors(); + return service.stop(watchIdentity(ctx), input.id); + }), + toModelOutput: output => toolOutput.json(output), +}); diff --git a/agent/tools/watch_update.ts b/agent/tools/watch_update.ts new file mode 100644 index 0000000..efaa2dc --- /dev/null +++ b/agent/tools/watch_update.ts @@ -0,0 +1,15 @@ +import { defineTool, toolOutput } from "eve/tools"; +import { z } from "zod"; +import { monitors } from "../../server/monitors/service"; +import { frequencyInput } from "../../server/monitors/config"; +import { watchIdentity, safeWatch } from "../lib/monitors"; + +export default defineTool({ + description: "Change an existing active watch's research frequency, from 1h to 30d. Faster checks increase scheduled usage. A pending_frequency is not confirmed; list watches to check completion. Does not create a replacement monitor.", + inputSchema: z.object({ id: z.string().uuid(), frequency: frequencyInput }), + execute: (input, ctx) => safeWatch(() => { + const owner = watchIdentity(ctx); + return monitors().update(owner, input.id, input.frequency); + }), + toModelOutput: output => toolOutput.json(output), +}); diff --git a/app/api/parallel/monitors/process/route.ts b/app/api/parallel/monitors/process/route.ts new file mode 100644 index 0000000..302d69f --- /dev/null +++ b/app/api/parallel/monitors/process/route.ts @@ -0,0 +1,7 @@ +import { processMonitors } from "../../../../../server/monitors/http"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; +export const maxDuration = 60; +export function GET(request: Request) { return processMonitors(request); } +export function POST(request: Request) { return processMonitors(request); } diff --git a/app/api/parallel/monitors/webhook/route.ts b/app/api/parallel/monitors/webhook/route.ts new file mode 100644 index 0000000..b9f87e3 --- /dev/null +++ b/app/api/parallel/monitors/webhook/route.ts @@ -0,0 +1,6 @@ +import { receiveWebhook } from "../../../../../server/monitors/http"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; +export const maxDuration = 60; +export function POST(request: Request) { return receiveWebhook(request); } diff --git a/docs/monitors.md b/docs/monitors.md new file mode 100644 index 0000000..211e035 --- /dev/null +++ b/docs/monitors.md @@ -0,0 +1,94 @@ +# Persistent public-web watches + +Ask Bell through an allowlisted, direct Linq conversation to watch a public topic. Bell creates a Parallel `event_stream` monitor using `lite`, without historical backfill. Research frequency defaults to daily and accepts whole hours, days, or weeks from `1h` through `30d`, including `6h` and `1w`. Faster checks increase scheduled usage; alerts still require detected changes. `watch_list` shows 50 watches per page with their frequencies and recent updates. `watch_update` changes an active watch's frequency; `watch_stop` cancels it permanently. Use a precise query describing the material changes you care about. Do not include private browser content or secrets. + +Parallel schedules the research. Bell stores ownership, events, and pending messages in PostgreSQL, then sends cited updates directly through Linq. Alerts do not use a model, Browserbase, or an Eve session, so the 24-hour session timeout does not end a watch. Existing search, browsing, and messaging still work without monitor configuration. + +## Setup + +Configure the existing `DATABASE_URL`, `PARALLEL_API_KEY`, `LINQ_API_KEY`, `LINQ_WEBHOOK_SECRET`, `BELL_LINQ_PHONE_NUMBER`, and `BELL_LINQ_ALLOWED_SENDERS`. Set `BELL_PUBLIC_URL` to your exact public HTTPS origin, with no trailing slash. The callback is `/api/parallel/monitors/webhook` and must be reachable by Parallel without interactive deployment authentication. + +Copy `PARALLEL_WEBHOOK_SECRET` from **Settings > Webhooks** in the Parallel organization associated with your API key. Do not generate a different secret locally. Set `CRON_SECRET` to an independent random secret of at least 32 bytes. Keep all credentials server-side. + +Apply the additive migration before deploying: + +```sh +psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -f migrations/004_parallel_monitors.sql +psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -f migrations/005_monitor_frequency.sql +``` + +Existing installations must apply `005_monitor_frequency.sql` before running the new code. It preserves existing watches as daily and is safe to reapply. Stop old workers before deploying the new code so they cannot consume the new update jobs. + +Monitor tables are independent of Link OAuth and can use the same database. Use a session-pooled connection. The worker uses bounded connection pools; delivery holds a watch row lock through the bounded Linq request to serialize with stopping. Frequency updates also hold the row lock through the bounded provider request, so they serialize with stopping. Other transactions are short; this does not make the rest of Bell's Eve workflow storage durable or multi-instance safe. + +Configure a scheduler to call `/api/parallel/monitors/process` every minute. Both GET and POST require `Authorization: Bearer `. For Vercel, add this to your deployment's `vercel.json` and configure `CRON_SECRET` in the environment: + +```json +{"crons":[{"path":"/api/parallel/monitors/process","schedule":"* * * * *"}]} +``` + +Vercel supplies the bearer header on its GET request. Hobby only supports daily cron jobs; minute-level delivery requires a suitable plan or an external scheduler making authenticated POST requests. Deploying the code alone does not activate processing. Verify successful scheduler invocations before creating a watch. The worker processes at most 30 jobs per invocation with a 20-second work-start budget, 8-second provider timeouts, and 60-second leases. The route allows 60 seconds for the final request and database cleanup. + +The webhook verifies Standard Webhooks signatures and timestamp freshness, then commits the callback before acknowledging. The worker fetches the actual events from Parallel. The saved sender and receiving line must still match Bell's allowlist before each outgoing send. Recipients cannot be supplied through tool arguments or callback metadata. + +## Listing watches + +`watch_list` returns one page of up to 50 watches in `watches`, plus a `next_cursor`. Open watches come first, then cancelled and failed ones, newest first within each group, so an older running watch stays on the first page no matter how many newer watches were cancelled after it. When `next_cursor` is not null there are older watches; pass it back as `cursor` to read the next page. The cursor is opaque and encodes a position in this order rather than an offset, so pages do not shift when a watch changes status between calls; an unusable cursor is rejected and the caller should list again without one. `status` accepts `all` (default), `open` for watches that are neither cancelled nor failed, and `closed` for those that are. Recent updates are additionally capped by a per-call content budget, reported as `updates_truncated` on each watch. + +## Status and recovery + +`active` means Parallel creation was confirmed. `dispatched` or `uncertain` means a create request may have succeeded. Bell never resends an uncertain create request. It searches the account's monitors for the exact opaque watch ID in metadata, including cancelled monitors. A crash just before sending can also leave an uncertain operation; inspect it instead of automatically creating a replacement. + +`frequency` is the last confirmed research interval for an active watch. `pending_frequency` is a queued or unresolved request, not proof that Parallel applied it. On creation, the returned interval is only confirmed when status is `active`. Equivalent intervals such as `7d` and `1w` are normalized. Frequency changes update the existing remote monitor and do not create a replacement. A lost response is retried with the same requested interval; the remote change may already have happened while local confirmation is pending. + +Only one frequency change may be pending per watch. Repeating the same request reports its status; a different request must wait until the prior change is confirmed or repaired. Failed/exhausted jobs appear in `problems` and use the repair procedure below. Stopping a watch clears pending changes and queues cancellation. Cadence settings do not change Bell's minute-level queue worker or hourly recovery scans. + +`watch_stop` records local suppression and queues remote cancellation. It can wait for an in-flight send; once stopping succeeds, no new alert is dispatched. If a lock wait times out, retry stopping. `cancel_pending` means remote cancellation is not yet confirmed, so billing may continue. Missing messaging/webhook configuration does not block stopping. Missing provider credentials leave cancellation pending until repaired. A message already being sent cannot be recalled. Removing a sender from the allowlist prevents sends but does not cancel their monitors; the operator must cancel them through Parallel or repair the local cancellation job. + +Linq retries reuse a frozen message body and idempotency key. `accepted` records provider acceptance, not device delivery or human receipt. Never generate a new key to work around a send error, including a 404 for an expired/deleted idempotent message. Repeated event IDs are deduplicated; similar events with different IDs can still appear. + +Callbacks are the primary delivery path. Hourly fallback scans recover missed events and health observations. Each scan starts at the newest page and deduplicates saved IDs. The inspected Parallel API returns a finite window of recent runs when completions are included. + +A scan that cannot finish inside the worker's work budget is not restarted from the newest page. The budget is checked between pages, so each claim fetches at least one whole page and then saves its place in the job payload: the next page cursor, the scan's start time and newest identity, and whether prior coverage was already found. The next run resumes from there, and the saved event IDs still suppress duplicate alerts for anything already stored. Saving progress is not a failure: it releases the retry attempt it consumed and clears the job's last error, so a long recovery no longer walks into the eight-attempt limit and stops there. Resuming is bounded: the scan only continues while its cursor keeps moving forward, and at most twenty resumes. Beyond that it is reported as an exceeded work budget like any other failure. The next scheduled check restarts a failed recovery job with a renewed twenty-resume allowance. Malformed-event failures restart from the newest events so corrected records can be rechecked; other failures retain their saved progress. Stored event IDs continue to prevent duplicate alerts. `recovery_gap` remains true when prior coverage cannot be found or recovery gives up. It does not claim that all missed history can be recovered. Completion/error records accept timezone offsets and normalize equivalent instants for recovery tracking. A malformed record does not prevent valid records from being processed; it leaves the scan incomplete with `invalid_monitor_event` in job problems and `recovery_gap=true`. Completion/error records update health without sending routine status messages. Event dates are detection/run dates, not necessarily the dates the underlying news occurred. + +Inspect `watch_list` for health, last scan, recovery gaps, and failed jobs. On the server, inspect queue age and failures: + +```sql +SELECT kind, count(*), min(created_at) AS oldest, bool_or(failed) AS has_failed +FROM bell_monitor_jobs GROUP BY kind; +SELECT id, watch_id, kind, attempts, last_error, failed +FROM bell_monitor_jobs WHERE last_error IS NOT NULL; +SELECT id, status, frequency, pending_frequency, health, health_at, last_scan_at, recovery_gap +FROM bell_monitors WHERE status <> 'cancelled'; +``` + +Transient jobs back off with jitter and stop after eight attempts. Provider authentication, validation, and other permanent failures stop sooner. A recovery scan that only ran out of work budget does not spend an attempt; it saves its place and resumes, as described above. Repair the underlying configuration or provider problem, then requeue only the inspected job, preserving its payload and key: + +```sql +UPDATE bell_monitor_jobs SET failed=false, attempts=0, next_at=now(), + lease_until=NULL, lease_token=NULL, last_error=NULL +WHERE id = '' AND (lease_until IS NULL OR lease_until < now()); +``` + +This retries reconciliation, event fetching, frequency updates, cancellation, or the same idempotent message. It never resets the watch's creation state. Resolve an uncertain watch by locating its `bell_watch_id` in Parallel; do not edit `remote_id` from unverified metadata. A failed cancellation remains pending until repaired. Keep the database backed up: deleting watch rows can orphan billed monitors, while deleting event records can allow duplicate alerts. + +Provision alerts for failed/aging jobs and missing scheduler invocations in your deployment. `bell_monitor_job_failed` logs contain job kind and retry state without secrets or message contents. No Bell-specific hosted alert is provisioned by this sample. Retained events and callbacks grow over time; this version does not automatically prune deduplication history. Restrict database access and account for storage usage. + +Before rollback, stop creating watches, suppress delivery, cancel owned monitors through Parallel, and confirm cancellation. Keep the database until uncertain creations and pending deliveries are resolved. Disabling the scheduler alone does not stop upstream billing. + +Parallel account settings may save public monitor inputs/outputs to personal memory when enabled. This integration does not opt into application memory or retrieve memories. Review that account setting for your deployment. + +## Local verification + +Point `TEST_MONITOR_DATABASE_URL` at a disposable PostgreSQL database, then run: + +```sh +pnpm exec vitest run tests/monitors.test.ts tests/monitor-http.test.ts tests/linq.test.ts +pnpm check +``` + +Database tests create isolated schemas and exercise concurrency, restart, uncertain creation, cancellation, and delivery retries. HTTP tests use signed fixtures and stub provider transports. These checks do not establish live Parallel or Linq delivery. Live delivery was intentionally not tested for this contribution. + +[Parallel Monitor API](https://docs.parallel.ai/api-reference/monitor/create-monitor) · [Monitor events](https://docs.parallel.ai/monitor-api/monitor-events) · [Webhook setup](https://docs.parallel.ai/resources/webhook-setup) · [Vercel Cron](https://vercel.com/docs/cron-jobs/manage-cron-jobs) + +[Update monitor frequency](https://docs.parallel.ai/api-reference/monitor/update-monitor) diff --git a/migrations/004_parallel_monitors.sql b/migrations/004_parallel_monitors.sql new file mode 100644 index 0000000..47791fd --- /dev/null +++ b/migrations/004_parallel_monitors.sql @@ -0,0 +1,46 @@ +CREATE TABLE IF NOT EXISTS bell_monitors ( + id uuid PRIMARY KEY, + issuer text NOT NULL, + owner_id text NOT NULL, + operation text NOT NULL, + query text NOT NULL, + destination jsonb NOT NULL, + remote_id text UNIQUE, + status text NOT NULL CHECK (status IN ('prepared','dispatched','uncertain','active','cancel_pending','cancelled','failed')), + stopped boolean NOT NULL DEFAULT false, + health text, + health_at timestamptz, + last_scan_at timestamptz, + scan_anchor text, + next_scan_at timestamptz NOT NULL DEFAULT now(), + recovery_gap boolean NOT NULL DEFAULT false, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (issuer, owner_id, operation) +); +CREATE TABLE IF NOT EXISTS bell_monitor_inbox ( + id text PRIMARY KEY, + payload jsonb NOT NULL, + received_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS bell_monitor_events ( + watch_id uuid NOT NULL REFERENCES bell_monitors(id), + event_id text NOT NULL, + payload jsonb NOT NULL, + received_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (watch_id, event_id) +); +CREATE TABLE IF NOT EXISTS bell_monitor_jobs ( + id text PRIMARY KEY, + watch_id uuid REFERENCES bell_monitors(id), + kind text NOT NULL CHECK (kind IN ('inbox','reconcile','sync','send','cancel')), + payload jsonb NOT NULL DEFAULT '{}', + next_at timestamptz NOT NULL DEFAULT now(), + lease_until timestamptz, + lease_token uuid, + attempts integer NOT NULL DEFAULT 0, + failed boolean NOT NULL DEFAULT false, + last_error text, + created_at timestamptz NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS bell_monitor_jobs_due ON bell_monitor_jobs(next_at) WHERE NOT failed; +CREATE INDEX IF NOT EXISTS bell_monitors_owner ON bell_monitors(issuer, owner_id, created_at DESC); diff --git a/migrations/005_monitor_frequency.sql b/migrations/005_monitor_frequency.sql new file mode 100644 index 0000000..80e9a88 --- /dev/null +++ b/migrations/005_monitor_frequency.sql @@ -0,0 +1,8 @@ +BEGIN; +ALTER TABLE bell_monitors ADD COLUMN IF NOT EXISTS creation_frequency text NOT NULL DEFAULT '1d'; +ALTER TABLE bell_monitors ADD COLUMN IF NOT EXISTS frequency text NOT NULL DEFAULT '1d'; +ALTER TABLE bell_monitors ADD COLUMN IF NOT EXISTS pending_frequency text; +ALTER TABLE bell_monitor_jobs DROP CONSTRAINT IF EXISTS bell_monitor_jobs_kind_check; +ALTER TABLE bell_monitor_jobs ADD CONSTRAINT bell_monitor_jobs_kind_check + CHECK (kind IN ('inbox','reconcile','sync','send','cancel','update')); +COMMIT; diff --git a/package.json b/package.json index 43f18ad..fafb213 100644 --- a/package.json +++ b/package.json @@ -35,6 +35,7 @@ "playwright-core": "1.62.1", "react": "19.3.0", "react-dom": "19.3.0", + "standardwebhooks": "1.1.1", "zod": "4.6.7" }, "devDependencies": { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e71bd2c..a010a4d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -49,6 +49,9 @@ importers: react-dom: specifier: 19.3.0 version: 19.3.0(react@19.3.0) + standardwebhooks: + specifier: 1.1.1 + version: 1.1.1 zod: specifier: 4.6.7 version: 4.6.7 @@ -802,6 +805,9 @@ packages: '@rtsao/scc@1.1.0': resolution: {integrity: sha512-zt6OdqaDoOnJ1ZYsCYGt9YmWzDXl4vQdKTyJev62gFhRGKdx7mcT54V9KIjg+d2wi9EXsPvAPKe7i7WjfVWB8g==} + '@stablelib/base64@1.0.1': + resolution: {integrity: sha512-1bnPQqSxSuc3Ii6MhBysoWCg58j97aUjuCSZrGSmDxNqtytIi0k8utUenAwTZN4V5mXXYGsVUI9zeBqy+jBOSQ==} + '@standard-schema/spec@1.1.0': resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} @@ -1583,6 +1589,9 @@ packages: fast-levenshtein@2.0.6: resolution: {integrity: sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==} + fast-sha256@1.3.0: + resolution: {integrity: sha512-n11RGP/lrWEFI/bWdygLxhI+pVeo1ZYIVwvvPkW7azl/rOy+F3HYRZ2K5zeE9mmkhQppyv9sQFx0JM9UabnpPQ==} + fastq@1.20.3: resolution: {integrity: sha512-XKv5nnLs6nLF71NgiKJLIZFLkPyIEuOselLG7ujZnGrRfQK8HpvY+WqKhAJUAdLomwVHErVS4LfxFlPq0/FTAw==} @@ -2434,6 +2443,9 @@ packages: stackback@0.0.2: resolution: {integrity: sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw==} + standardwebhooks@1.1.1: + resolution: {integrity: sha512-bCbX9ZEyFkWPsRz7Bl3NuQUJohmwGSev/yhr7vhaGPlc4AfIrspIRa6cPTBuI1ItmrTDJ4d/S2hCsfe4+vQGnQ==} + std-env@4.3.0: resolution: {integrity: sha512-OtU/EgQ1kIm5KwqQpBC6ZEMXrZRui11w8zgfTWp8cdO9B8OaPsbA8bTHO2P+HNo1VlUTGMVBwPhydu6poeXiag==} @@ -3313,6 +3325,8 @@ snapshots: '@rtsao/scc@1.1.0': {} + '@stablelib/base64@1.0.1': {} + '@standard-schema/spec@1.1.0': {} '@stripe/link-integrations-eve@0.3.4(eve@0.79.1(@opentelemetry/api@1.9.1)(ai@7.0.137(zod@4.6.7))(microsandbox@0.5.10))': @@ -4247,6 +4261,8 @@ snapshots: fast-levenshtein@2.0.6: {} + fast-sha256@1.3.0: {} + fastq@1.20.3: dependencies: reusify: 1.1.0 @@ -5125,6 +5141,11 @@ snapshots: stackback@0.0.2: {} + standardwebhooks@1.1.1: + dependencies: + '@stablelib/base64': 1.0.1 + fast-sha256: 1.3.0 + std-env@4.3.0: {} stop-iteration-iterator@1.1.0: diff --git a/server/monitors/config.ts b/server/monitors/config.ts new file mode 100644 index 0000000..26be3ef --- /dev/null +++ b/server/monitors/config.ts @@ -0,0 +1,42 @@ +import { z } from "zod"; +import { publicUrl } from "../parallel"; + +export interface Owner { issuer: string; id: string } +export interface Destination { chat: string; sender: string; line: string } +export const queryInput = z.string().trim().min(1).max(2_000); +const phone = z.string().regex(/^\+[1-9]\d{1,14}$/); +export const destinationSchema = z.object({ chat: z.string().min(1).max(256), sender: phone, line: phone }); +export function allowed(destination: Destination): boolean { + const line = phone.safeParse(process.env.BELL_LINQ_PHONE_NUMBER?.trim()); + const senders = z.array(phone).nonempty().safeParse((process.env.BELL_LINQ_ALLOWED_SENDERS ?? "").split(",").map(s => s.trim())); + return line.success && senders.success && line.data === destination.line && senders.data.includes(destination.sender); +} +export function required(name: string): string { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Watches unavailable: configure ${name}.`); + return value; +} +export function creationOrigin(): string { + for (const name of ["DATABASE_URL", "PARALLEL_API_KEY", "PARALLEL_WEBHOOK_SECRET", "CRON_SECRET", "LINQ_API_KEY"]) required(name); + const origin = required("BELL_PUBLIC_URL"), url = new URL(origin); + if (!publicUrl.safeParse(origin).success || url.protocol !== "https:" || url.origin !== origin || !url.hostname.includes(".") || /(^|\.)(localhost|local|internal|test|invalid)$/.test(url.hostname)) throw new Error("Watches require a public HTTPS BELL_PUBLIC_URL origin."); + return origin; +} +export function assertOwner(owner: Owner): void { + if (owner.issuer !== "linq" || !/^linq:.{1,507}$/.test(owner.id)) throw new Error("Watches require an authenticated Linq sender."); +} + +export const frequencyInput = z.string().regex(/^[1-9]\d{0,2}[hdw]$/, "Use whole hours, days, or weeks, between 1h and 30d.").refine(value => { + const hours = Number(value.slice(0, -1)) * ({ h: 1, d: 24, w: 168 }[value.slice(-1)] ?? 0); + return hours >= 1 && hours <= 720; +}, "Frequency must be between 1h and 30d.").transform(value => { + const hours = Number(value.slice(0, -1)) * ({ h: 1, d: 24, w: 168 }[value.slice(-1)]!); + return hours % 168 === 0 ? `${hours / 168}w` : hours % 24 === 0 ? `${hours / 24}d` : `${hours}h`; +}); + +export const monitorTimestamp = z.string().datetime({ offset: true }).transform(value => { + // Preserve sub-millisecond precision while giving equivalent instants one recovery identity. + const fraction = (value.match(/\.(\d+)/)?.[1] ?? "").replace(/0+$/, ""); + const seconds = new Date(value.replace(/\.\d+/, "")).toISOString().slice(0, 19); + return `${seconds}${fraction ? `.${fraction}` : ""}Z`; +}); diff --git a/server/monitors/http.ts b/server/monitors/http.ts new file mode 100644 index 0000000..b1f349d --- /dev/null +++ b/server/monitors/http.ts @@ -0,0 +1,26 @@ +import { timingSafeEqual } from "node:crypto"; +import { WebhookVerificationError } from "standardwebhooks"; +import { ZodError } from "zod"; +import { monitors, type MonitorService } from "./service"; +import { readBounded } from "./provider"; + +export async function receiveWebhook(request: Request, service: () => MonitorService = monitors): Promise { + if (!process.env.PARALLEL_WEBHOOK_SECRET?.trim()) return new Response("Unavailable", { status: 503 }); + if (Number(request.headers.get("content-length")) > 65_536) return new Response("Too large", { status: 413 }); + try { + const body = await readBounded(new Response(request.body), 65_536); + await service().ingest(body, Object.fromEntries(request.headers)); + return new Response(null, { status: 204 }); + } catch (error) { + if (error instanceof WebhookVerificationError) return new Response("Invalid signature", { status: 401 }); + if (error instanceof ZodError || error instanceof SyntaxError) return new Response("Invalid event", { status: 400 }); + return new Response("Unavailable", { status: 503 }); + } +} +export async function processMonitors(request: Request, service: () => MonitorService = monitors): Promise { + const secret = process.env.CRON_SECRET?.trim(), supplied = Buffer.from(request.headers.get("authorization") ?? ""); + const expected = Buffer.from(`Bearer ${secret}`); + if (!secret || expected.length !== supplied.length || !timingSafeEqual(expected, supplied)) return new Response("Unauthorized", { status: 401 }); + try { return Response.json(await service().process(), { headers: { "Cache-Control": "no-store" } }); } + catch { return new Response("Unavailable", { status: 503 }); } +} diff --git a/server/monitors/provider.ts b/server/monitors/provider.ts new file mode 100644 index 0000000..1cfd619 --- /dev/null +++ b/server/monitors/provider.ts @@ -0,0 +1,73 @@ +import { z } from "zod"; +import { publicUrl } from "../parallel"; +import { type Destination, required, frequencyInput, monitorTimestamp } from "./config"; + +const remoteSchema = z.object({ monitor_id: z.string().min(1), status: z.enum(["active", "cancelled"]), frequency: frequencyInput, metadata: z.record(z.string(), z.string()).nullish() }); +const citation = z.object({ url: publicUrl.nullable().catch(null) }); +export const eventSchema = z.discriminatedUnion("event_type", [ + z.object({ event_type: z.literal("event_stream"), event_id: z.string(), event_group_id: z.string(), event_date: z.string().nullish(), output: z.object({ type: z.enum(["text", "json"]), content: z.union([z.string(), z.record(z.string(), z.unknown())]), basis: z.array(z.object({ citations: z.array(citation).default([]) })).default([]) }) }), + z.object({ event_type: z.literal("completion"), timestamp: monitorTimestamp }), + z.object({ event_type: z.literal("error"), timestamp: monitorTimestamp, error_message: z.string() }), +]); +export type MonitorEvent = z.infer; +export interface RemoteMonitor { monitor_id: string; status: string; frequency: string; metadata?: Record | null } +export interface MonitorProvider { + create(query: string, watchId: string, origin: string, frequency?: string): Promise; + list(cursor?: string): Promise<{ monitors: RemoteMonitor[]; next_cursor?: string | null }>; + events(id: string, group?: string, cursor?: string): Promise<{ events: unknown[]; next_cursor?: string | null }>; + cancel(id: string): Promise; + update(id: string, frequency: string): Promise; +} +export class ProviderError extends Error { + constructor(readonly status: number, readonly retryAfter = 0) { super(`Provider HTTP ${status}`); } +} +export async function readBounded(response: Response, maxBytes = 1_048_576): Promise { + const reader = response.body?.getReader(); + if (!reader) return ""; + const chunks: Uint8Array[] = []; let bytes = 0; + try { + while (true) { + const { done, value } = await reader.read(); if (done) break; + bytes += value.byteLength; + if (bytes > maxBytes) throw new Error("Response exceeds size limit"); + chunks.push(value); + } + return Buffer.concat(chunks).toString("utf8"); + } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); } +} +export async function request(url: string, key: string, method = "GET", body?: unknown, bearer = false): Promise { + const response = await fetch(url, { method, headers: { "Content-Type": "application/json", ...(bearer ? { Authorization: `Bearer ${key}` } : { "x-api-key": key }) }, body: body === undefined ? undefined : JSON.stringify(body), signal: AbortSignal.timeout(8_000), redirect: "error", cache: "no-store" }); + if (!response.ok) { + await response.body?.cancel(); + const value = response.headers.get("retry-after"), seconds = Number(value); + const delay = value ? (Number.isFinite(seconds) ? seconds * 1000 : Date.parse(value) - Date.now()) : 0; + throw new ProviderError(response.status, Math.max(0, Math.min(delay || 0, 86_400_000))); + } + const text = await readBounded(response); + return text ? JSON.parse(text) : null; +} +const path = (id: string) => `/monitors/${encodeURIComponent(id)}`; +const call = (endpoint: string, method = "GET", body?: unknown) => request(`https://api.parallel.ai/v1${endpoint}`, required("PARALLEL_API_KEY"), method, body); +export const parallelMonitors: MonitorProvider = { + async create(query, watchId, origin, frequency = "1d") { + return remoteSchema.parse(await call("/monitors", "POST", { type: "event_stream", frequency: frequencyInput.parse(frequency), processor: "lite", settings: { query, include_backfill: false }, metadata: { bell_watch_id: watchId }, webhook: { url: `${origin}/api/parallel/monitors/webhook`, event_types: ["monitor.event.detected", "monitor.execution.completed", "monitor.execution.failed"] } })); + }, + async list(cursor) { + const params = new URLSearchParams({ type: "event_stream", limit: "100" }); + params.append("status", "active"); params.append("status", "cancelled"); if (cursor) params.set("cursor", cursor); + return z.object({ monitors: z.array(remoteSchema), next_cursor: z.string().nullish() }).parse(await call(`/monitors?${params}`)); + }, + async events(id, group, cursor) { + const params = new URLSearchParams(group ? { event_group_id: group } : { include_completions: "true", limit: "100" }); + if (cursor) params.set("cursor", cursor); + return z.object({ events: z.array(z.unknown()), next_cursor: z.string().nullish() }).parse(await call(`${path(id)}/events?${params}`)); + }, + async update(id, frequency) { + return remoteSchema.parse(await call(`${path(id)}/update`, "POST", { frequency: frequencyInput.parse(frequency) })); + }, + async cancel(id) { await call(`${path(id)}/cancel`, "POST"); }, +}; +export async function sendLinq(destination: Destination, text: string, key: string): Promise { + const response = await request(`https://api.linqapp.com/api/partner/v3/chats/${encodeURIComponent(destination.chat)}/messages`, required("LINQ_API_KEY"), "POST", { message: { parts: [{ type: "text", value: text }], idempotency_key: key } }, true); + return z.object({ message: z.object({ id: z.string() }) }).parse(response).message.id; +} diff --git a/server/monitors/service.ts b/server/monitors/service.ts new file mode 100644 index 0000000..fd22445 --- /dev/null +++ b/server/monitors/service.ts @@ -0,0 +1,232 @@ +import { createHash } from "node:crypto"; +import { Webhook } from "standardwebhooks"; +import { z } from "zod"; +import { allowed, assertOwner, creationOrigin, destinationSchema, type Destination, type Owner, queryInput, required, frequencyInput, monitorTimestamp } from "./config"; +import { eventSchema, type MonitorProvider, parallelMonitors, ProviderError, sendLinq } from "./provider"; +import { type Job, type Watch, MonitorStore, monitorStore, watchFilter } from "./store"; + +const callbackTimestamp = monitorTimestamp; +class InvalidMonitorEvent extends Error {} +const callbackData = z.object({ monitor_id: z.string(), metadata: z.record(z.string(), z.string()).nullish() }); +const callbackSchema = z.discriminatedUnion("type", [ + z.object({ type: z.literal("monitor.event.detected"), timestamp: callbackTimestamp, data: callbackData.extend({ event: z.object({ event_group_id: z.string() }) }) }), + z.object({ type: z.literal("monitor.execution.completed"), timestamp: callbackTimestamp, data: callbackData.extend({ event: z.object({ type: z.literal("completion"), monitor_ts: z.string() }) }) }), + z.object({ type: z.literal("monitor.execution.failed"), timestamp: callbackTimestamp, data: callbackData.extend({ event: z.object({ type: z.literal("error"), error: z.string(), id: z.string(), date: callbackTimestamp }) }) }), +]); +const hash = (value: string) => createHash("sha256").update(value).digest("hex"); +// Saved place of a recovery scan, so a run stopped by the work budget resumes instead of refetching from the newest page. +const syncProgress = z.object({ + group: z.string().optional(), cursor: z.string().optional(), started: z.string().datetime().optional(), anchor: z.string().optional(), + anchor_seen: z.boolean().default(false), runs: z.array(z.string()).default([]), invalid: z.boolean().default(false), resumes: z.number().int().min(0).default(0), +}); +const resumeLimit = 20; +export class MonitorService { + constructor(readonly store: MonitorStore, readonly provider: MonitorProvider = parallelMonitors, readonly send: (destination: Destination, text: string, key: string) => Promise = sendLinq, readonly budget = 20_000) {} + async create(owner: Owner, destination: Destination, operation: string, input: string, cadence = "1d") { + assertOwner(owner); destinationSchema.parse(destination); + if (!allowed(destination)) throw new Error("This Linq destination is no longer allowed."); + const origin = creationOrigin(), query = queryInput.parse(input), frequency = frequencyInput.parse(cadence); + const watch = await this.store.prepare(owner, destination, operation, query, frequency); + if (await this.store.dispatch(watch.id)) { + try { + const remote = await this.provider.create(query, watch.id, origin, frequency); + await this.store.bind(watch.id, remote.monitor_id, remote.status === "cancelled", remote.frequency); + } catch (error) { + const rejected = error instanceof ProviderError && [400,401,403,422].includes(error.status); + await this.store.pool.query("UPDATE bell_monitors SET status=$2 WHERE id=$1 AND status='dispatched'", [watch.id, rejected ? "failed" : "uncertain"]); + await this.store.pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE id=$1", [`reconcile:${watch.id}`]); + } + } + const result = await this.store.get(watch.id); + return { id: result.id, status: result.status, frequency: result.frequency, pending_frequency: result.pending_frequency, processor: "lite" }; + } + async list(owner: Owner, options: { status?: string; cursor?: string } = {}) { + assertOwner(owner); + const status = watchFilter.parse(options.status ?? "all"); + const page = await this.store.list(owner, status, options.cursor); + let remaining = 12_000; + const watches = page.watches.map(w => { + const updates: unknown[] = []; + for (const raw of w.updates) { + if (remaining <= 0) break; + const event = eventSchema.parse(raw); + if (event.event_type !== "event_stream") continue; + const text = typeof event.output.content === "string" ? event.output.content : JSON.stringify(event.output.content); + const content = text.slice(0, Math.min(600, remaining)); + remaining -= content.length; + const citations: string[] = []; + for (const url of [...new Set(event.output.basis.flatMap(b => b.citations.flatMap(c => c.url ? [c.url] : [])))].slice(0, 3)) { + if (url.length <= remaining) { citations.push(url); remaining -= url.length; } + } + updates.push({ event_id: event.event_id, event_date: event.event_date, content, citations, truncated: content.length < text.length }); + } + return { id: w.id, query: w.query.slice(0, 400), status: w.status, frequency: w.frequency, pending_frequency: w.pending_frequency, health: w.health, health_at: w.health_at, recovery_gap: w.recovery_gap, last_scan_at: w.last_scan_at, updates, updates_truncated: updates.length < w.updates.length, problems: w.problems }; + }); + return { watches, next_cursor: page.next_cursor }; + } + async update(owner: Owner, id: string, cadence: string) { + assertOwner(owner); z.string().uuid().parse(id); + const watch = await this.store.requestFrequency(owner, id, frequencyInput.parse(cadence)); + return { id: watch.id, status: watch.status, frequency: watch.frequency, pending_frequency: watch.pending_frequency }; + } + async stop(owner: Owner, id: string) { + assertOwner(owner); z.string().uuid().parse(id); + await this.store.stop(owner, id); + return { id, status: (await this.store.get(id)).status }; + } + async ingest(body: string, headers: Record) { + if (Buffer.byteLength(body) > 65_536) throw new Error("Webhook too large"); + const parsed = new Webhook(required("PARALLEL_WEBHOOK_SECRET")).verify(body, headers); + const event = callbackSchema.parse(parsed), id = headers["webhook-id"]; + if (!id || id.length > 256) throw new Error("Invalid webhook identity"); + await this.store.transaction(async client => { + const inserted = await client.query("INSERT INTO bell_monitor_inbox(id,payload) VALUES ($1,$2) ON CONFLICT DO NOTHING RETURNING id", [id, event]); + if (inserted.rowCount) await this.store.enqueue("inbox", `inbox:${id}`, null, { id }, client); + }); + } + async process() { + const deadline = Date.now() + this.budget; + await this.store.schedule(); + let processed = 0; + while (processed < 30 && Date.now() < deadline) { + const job = await this.store.claim(); if (!job) break; + try { + const resume = await this.run(job, deadline); + if (resume) await this.store.defer(job, resume); else await this.store.finish(job); + } + catch (error) { + const provider = error instanceof ProviderError ? error : undefined; + const terminal = !!provider && [400,401,402,403,404,422].includes(provider.status); + const delay = Math.max(provider?.retryAfter ?? 0, Math.min(3_600_000, 1000 * 2 ** job.attempts) + Math.floor(Math.random() * 1000)); + await this.store.retry(job, provider ? `provider_http_${provider.status}` : error instanceof InvalidMonitorEvent ? "invalid_monitor_event" : "processing_failed", terminal, delay); + console.warn("bell_monitor_job_failed", { kind: job.kind, attempt: job.attempts, terminal: terminal || job.attempts >= 8 }); + } + processed++; + } + return { processed }; + } + /** Resolves to a saved payload when the work budget ran out mid-scan, so the next run resumes from there. */ + private async run(job: Job, deadline: number): Promise | undefined> { + if (job.kind === "inbox") { + const payload = (await this.store.pool.query("SELECT payload FROM bell_monitor_inbox WHERE id=$1", [job.payload.id])).rows[0]?.payload; + const event = callbackSchema.parse(payload); + let watch = (await this.store.pool.query("SELECT * FROM bell_monitors WHERE remote_id=$1", [event.data.monitor_id])).rows[0]; + if (!watch) { + const id = event.data.metadata?.bell_watch_id; + if (!id || !z.string().uuid().safeParse(id).success) return; + watch = (await this.store.pool.query("SELECT * FROM bell_monitors WHERE id=$1 AND status IN ('dispatched','uncertain','cancel_pending')", [id])).rows[0]; + if (!watch) return; + // Early callbacks wait for API-backed reconciliation; metadata is not ownership proof. + throw new Error("Monitor identity not reconciled yet"); + } + await this.store.health(watch.id, event.type === "monitor.execution.failed" ? "failed" : "completed", event.timestamp); + if (!watch.stopped && watch.status === "active" && event.type === "monitor.event.detected") { + await this.store.enqueue("sync", `group:${watch.id}:${hash(event.data.event.event_group_id)}`, watch.id, { group: event.data.event.event_group_id }); + } + return; + } + const watch = await this.store.get(job.watch_id!); + if (job.kind === "reconcile") { + if (watch.remote_id || ["failed", "cancelled"].includes(watch.status)) return; + let cursor = typeof job.payload.cursor === "string" ? job.payload.cursor : undefined; + for (let page = 0; page < 5 && Date.now() < deadline; page++) { + const result = await this.provider.list(cursor); + const matches = result.monitors.filter(m => m.metadata?.bell_watch_id === watch.id); + if (matches.length > 1) throw new Error("Multiple remote matches require operator reconciliation"); + if (matches.length) { await this.store.bind(watch.id, matches[0].monitor_id, matches[0].status === "cancelled", matches[0].frequency); return; } + cursor = result.next_cursor ?? undefined; + if (!cursor) break; + } + await this.store.pool.query("UPDATE bell_monitor_jobs SET payload=$3 WHERE id=$1 AND lease_token=$2", [job.id, job.lease_token, cursor ? { cursor } : {}]); + throw new Error("Creation remains uncertain"); + } + if (job.kind === "cancel") { + if (!watch.remote_id) throw new Error("Creation unresolved"); + await this.provider.cancel(watch.remote_id); + await this.store.pool.query("UPDATE bell_monitors SET status='cancelled' WHERE id=$1", [watch.id]); return; + } + if (watch.stopped || watch.status !== "active") return; + if (job.kind === "update") { + const frequency = frequencyInput.parse(job.payload.frequency); + await this.store.applyFrequency(watch.id, frequency, async id => { + const remote = await this.provider.update(id, frequency); + if (remote.monitor_id !== id || remote.status !== "active" || remote.frequency !== frequency) throw new Error("Monitor frequency not confirmed"); + return remote.frequency; + }); + return; + } + if (job.kind === "send") { + const payload = z.object({ text: z.string(), key: z.string(), eventId: z.string() }).parse(job.payload); + await this.store.deliver(watch.id, payload.eventId, async current => { + if (!allowed(current.destination)) throw new ProviderError(403); + return this.send(current.destination, payload.text, payload.key); + }); + return; + } + if (job.kind === "sync") { + if (!watch.remote_id) throw new Error("Missing remote monitor"); + const saved = syncProgress.parse(job.payload); + const group = saved.group; + const started = saved.started ?? new Date().toISOString(); + let cursor = saved.cursor, anchor = saved.anchor, previousAnchorSeen = saved.anchor_seen, invalidEvents = saved.invalid; + const runs = new Set(saved.runs); + const prior = watch.scan_anchor?.match(/^(completion|error):(.*)$/); + const priorTimestamp = prior ? monitorTimestamp.safeParse(prior[2]) : undefined; + const previousAnchor = prior && priorTimestamp?.success ? `${prior[1]}:${priorTimestamp.data}` : watch.scan_anchor; + // Each claim fetches at least one page, then stops on a page boundary so the saved cursor never splits a page. + for (let page = 0; page < 10; page++) { + if (page && Date.now() >= deadline) break; + const result = await this.provider.events(watch.remote_id, group, cursor); + const events = result.events.flatMap(raw => { + const parsed = eventSchema.safeParse(raw); + if (parsed.success) return [parsed.data]; + invalidEvents = true; + return []; + }); + const ids = events.flatMap(event => event.event_type === "event_stream" ? [event.event_id] : []); + const known = new Set((await this.store.pool.query<{ event_id: string }>("SELECT event_id FROM bell_monitor_events WHERE watch_id=$1 AND event_id=ANY($2::text[])", [watch.id, ids])).rows.map(row => row.event_id)); + for (const event of events) { + const identity = event.event_type === "event_stream" ? event.event_id : `${event.event_type}:${event.timestamp}`; + anchor ??= identity; + if (identity === previousAnchor) previousAnchorSeen = true; + if (runs.size < 100) runs.add(event.event_type === "event_stream" ? event.event_group_id : identity); + if (event.event_type !== "event_stream" || !known.has(event.event_id)) await this.event(watch, event); + } + const next = result.next_cursor ?? undefined; + if (!next) { + if (invalidEvents) { + await this.store.pool.query("UPDATE bell_monitors SET recovery_gap=true WHERE id=$1", [watch.id]); + throw new InvalidMonitorEvent("Invalid monitor event in recovery scan"); + } + if (!group) { + const gap = watch.scan_anchor ? !previousAnchorSeen : runs.size >= 100; + await this.store.pool.query("UPDATE bell_monitors SET last_scan_at=$2,scan_anchor=COALESCE($3,scan_anchor),recovery_gap=recovery_gap OR $4 WHERE id=$1", [watch.id, started, anchor ?? null, gap]); + } + return; + } + cursor = next; + } + // Only resume while the scan keeps moving forward, so a stalled cursor still surfaces as a gap instead of looping. + if (cursor && cursor !== saved.cursor && saved.resumes < resumeLimit) { + return { group, cursor, started, anchor, anchor_seen: previousAnchorSeen, runs: [...runs], invalid: invalidEvents, resumes: saved.resumes + 1 }; + } + await this.store.pool.query("UPDATE bell_monitors SET recovery_gap=true WHERE id=$1", [watch.id]); + throw new Error("Recovery scan exceeded work budget"); + } + } + private async event(watch: Watch, raw: unknown) { + const event = eventSchema.parse(raw); + if (event.event_type !== "event_stream") { + await this.store.health(watch.id, event.event_type === "completion" ? "completed" : "failed", event.timestamp); return; + } + const content = typeof event.output.content === "string" ? event.output.content : JSON.stringify(event.output.content); + const urls = [...new Set(event.output.basis.flatMap(b => b.citations.map(c => c.url).filter((u): u is string => !!u)))].slice(0, 3); + const text = `Watch update: ${watch.query.slice(0, 120)}\n${content.slice(0, 1200)}${content.length > 1200 ? "…" : ""}\n${urls.join("\n")}`; + const key = `bell-watch-${hash(`${watch.id}:${event.event_id}`)}`; + await this.store.transaction(async client => { + const inserted = await client.query("INSERT INTO bell_monitor_events (watch_id,event_id,payload) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING RETURNING event_id", [watch.id, event.event_id, event]); + if (inserted.rowCount) await this.store.enqueue("send", key, watch.id, { text, key, eventId: event.event_id }, client); + }); + } +} +export function monitors() { return new MonitorService(monitorStore()); } diff --git a/server/monitors/store.ts b/server/monitors/store.ts new file mode 100644 index 0000000..d1db9dc --- /dev/null +++ b/server/monitors/store.ts @@ -0,0 +1,162 @@ +import { randomUUID } from "node:crypto"; +import { Pool, type PoolClient } from "pg"; +import { z } from "zod"; +import { type Destination, type Owner, required } from "./config"; + +export interface Watch { + id: string; issuer: string; owner_id: string; operation: string; query: string; creation_frequency: string; frequency: string; pending_frequency: string | null; destination: Destination; + remote_id: string | null; status: string; stopped: boolean; health: string | null; health_at: Date | null; + created_at: Date; last_scan_at: Date | null; scan_anchor: string | null; recovery_gap: boolean; +} +export interface Job { id: string; watch_id: string | null; kind: string; payload: Record; attempts: number; lease_token: string } +export type ListedWatch = Watch & { updates: unknown[]; problems: unknown[] }; +/** `open` keeps watches that may still run or still need repair; `closed` is everything already cancelled or failed. */ +export const watchFilter = z.enum(["all", "open", "closed"]); +export type WatchFilter = z.infer; +export const watchPageSize = 50; +const closedStatuses = "'cancelled','failed'"; +// Keyset page marker for the (rank, created_at DESC, id DESC) order, opaque to the agent so the ordering can change. +const cursorSchema = z.object({ rank: z.int().min(0).max(1), created_at: z.string().datetime(), id: z.uuid() }); +const encodeCursor = (value: z.infer) => Buffer.from(JSON.stringify(value)).toString("base64url"); +function decodeCursor(cursor: string) { + try { + if (cursor.length > 512) throw new Error("Cursor too long"); + return cursorSchema.parse(JSON.parse(Buffer.from(cursor, "base64url").toString("utf8"))); + } catch { + throw new Error("Watch operation received an unusable page cursor. List watches again without a cursor."); + } +} +export class MonitorStore { + constructor(readonly pool: Pool) {} + async transaction(fn: (client: PoolClient) => Promise): Promise { + const client = await this.pool.connect(); + try { await client.query("BEGIN"); const value = await fn(client); await client.query("COMMIT"); return value; } + catch (error) { await client.query("ROLLBACK"); throw error; } + finally { client.release(); } + } + async prepare(owner: Owner, destination: Destination, operation: string, query: string, frequency = "1d"): Promise { + await this.pool.query("INSERT INTO bell_monitors (id,issuer,owner_id,operation,query,destination,status,creation_frequency,frequency) VALUES ($1,$2,$3,$4,$5,$6,'prepared',$7,$7) ON CONFLICT (issuer,owner_id,operation) DO NOTHING", [randomUUID(), owner.issuer, owner.id, operation, query, destination, frequency]); + const watch = (await this.pool.query("SELECT * FROM bell_monitors WHERE issuer=$1 AND owner_id=$2 AND operation=$3", [owner.issuer, owner.id, operation])).rows[0]; + if (watch.creation_frequency !== frequency || watch.query !== query || watch.destination.chat !== destination.chat || watch.destination.sender !== destination.sender || watch.destination.line !== destination.line) throw new Error("Watch operation has different input."); + return watch; + } + async get(id: string): Promise { + const watch = (await this.pool.query("SELECT * FROM bell_monitors WHERE id=$1", [id])).rows[0]; + if (!watch) throw new Error("Watch not found."); return watch; + } + async enqueue(kind: string, id: string, watchId: string | null, payload: Record = {}, client: Pool | PoolClient = this.pool) { + await client.query("INSERT INTO bell_monitor_jobs (id,watch_id,kind,payload) VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING", [id, watchId, kind, payload]); + } + async dispatch(id: string): Promise { + return this.transaction(async client => { + const result = await client.query("UPDATE bell_monitors SET status='dispatched' WHERE id=$1 AND status='prepared' AND NOT stopped RETURNING id", [id]); + if (!result.rowCount) return false; + await this.enqueue("reconcile", `reconcile:${id}`, id, {}, client); + await client.query("UPDATE bell_monitor_jobs SET next_at=now()+interval '1 minute' WHERE id=$1", [`reconcile:${id}`]); + return true; + }); + } + async bind(id: string, remoteId: string, cancelled = false, frequency = "1d") { + await this.transaction(async client => { + const watch = (await client.query("SELECT * FROM bell_monitors WHERE id=$1 FOR UPDATE", [id])).rows[0]; + if (watch.remote_id) { + if (watch.remote_id !== remoteId) throw new Error("Conflicting monitor identity"); + return; + } + await client.query("UPDATE bell_monitors SET remote_id=$2,status=$3,frequency=$4 WHERE id=$1", [id, remoteId, cancelled ? "cancelled" : watch.stopped ? "cancel_pending" : "active", frequency]); + if (watch.stopped && !cancelled) await this.enqueue("cancel", `cancel:${id}`, id, {}, client); + }); + } + async stop(owner: Owner, id: string) { + await this.transaction(async client => { + const watch = (await client.query("SELECT * FROM bell_monitors WHERE id=$1 AND issuer=$2 AND owner_id=$3 FOR UPDATE", [id, owner.issuer, owner.id])).rows[0]; + if (!watch) throw new Error("Watch not found."); + const localOnly = watch.status === "prepared" || watch.status === "failed"; + await client.query("UPDATE bell_monitors SET stopped=true,pending_frequency=NULL,status=$2 WHERE id=$1", [id, watch.status === "cancelled" || localOnly ? "cancelled" : "cancel_pending"]); + await client.query("DELETE FROM bell_monitor_jobs WHERE watch_id=$1 AND kind IN ('send','update') AND lease_until IS NULL", [id]); + if (watch.remote_id && watch.status !== "cancelled") await this.enqueue("cancel", `cancel:${id}`, id, {}, client); + }); + } + async requestFrequency(owner: Owner, id: string, frequency: string): Promise { + return this.transaction(async client => { + const watch = (await client.query("SELECT * FROM bell_monitors WHERE id=$1 AND issuer=$2 AND owner_id=$3 FOR UPDATE", [id, owner.issuer, owner.id])).rows[0]; + if (!watch) throw new Error("Watch not found."); + if (watch.stopped || watch.status !== "active" || !watch.remote_id) throw new Error("Watch operation requires an active watch."); + if (watch.pending_frequency) { + if (watch.pending_frequency !== frequency) throw new Error("Watch operation has a pending frequency change. List watches before requesting another change."); + return watch; + } + if (watch.frequency === frequency) return watch; + await client.query("UPDATE bell_monitors SET pending_frequency=$2 WHERE id=$1", [id, frequency]); + await this.enqueue("update", `update:${id}:${randomUUID()}`, id, { frequency }, client); + return { ...watch, pending_frequency: frequency }; + }); + } + async applyFrequency(id: string, frequency: string, update: (remoteId: string) => Promise) { + await this.transaction(async client => { + const watch = (await client.query("SELECT * FROM bell_monitors WHERE id=$1 FOR UPDATE", [id])).rows[0]; + if (!watch || watch.stopped || watch.status !== "active" || !watch.remote_id || watch.pending_frequency !== frequency) return; + const confirmed = await update(watch.remote_id); + await client.query("UPDATE bell_monitors SET frequency=$2,pending_frequency=NULL WHERE id=$1", [id, confirmed]); + }); + } + async deliver(id: string, eventId: string, send: (watch: Watch) => Promise) { + await this.transaction(async client => { + // Share stop's row lock through dispatch so suppression cannot race a stale read. + const watch = (await client.query("SELECT * FROM bell_monitors WHERE id=$1 FOR UPDATE", [id])).rows[0]; + if (!watch || watch.stopped || watch.status !== "active") return; + const accepted = await send(watch); + await client.query("UPDATE bell_monitor_events SET payload=jsonb_set(payload,'{delivery}', $3::jsonb) WHERE watch_id=$1 AND event_id=$2", [id, eventId, JSON.stringify({ status: "accepted", message_id: accepted })]); + }); + } + async list(owner: Owner, filter: WatchFilter = "all", cursor?: string): Promise<{ watches: ListedWatch[]; next_cursor: string | null }> { + const from = cursor ? decodeCursor(cursor) : undefined; + const rows = (await this.pool.query(`WITH scoped AS ( + SELECT w.*, (CASE WHEN w.status IN (${closedStatuses}) THEN 1 ELSE 0 END) AS rank FROM bell_monitors w WHERE issuer=$1 AND owner_id=$2) + SELECT w.*, to_char(w.created_at AT TIME ZONE 'UTC','YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cursor_created_at, COALESCE((SELECT jsonb_agg(e.payload) FROM (SELECT payload FROM bell_monitor_events WHERE watch_id=w.id ORDER BY received_at DESC LIMIT 5) e),'[]') AS updates, + COALESCE((SELECT jsonb_agg(jsonb_build_object('kind',kind,'error',last_error,'failed',failed)) FROM bell_monitor_jobs WHERE watch_id=w.id AND last_error IS NOT NULL),'[]') AS problems + FROM scoped w WHERE ($3::int IS NULL OR w.rank=$3) + AND ($4::int IS NULL OR w.rank>$4 OR (w.rank=$4 AND (w.created_at<$5::timestamptz OR (w.created_at=$5::timestamptz AND w.id<$6::uuid)))) + ORDER BY w.rank, w.created_at DESC, w.id DESC LIMIT ${watchPageSize + 1}`, + [owner.issuer, owner.id, filter === "all" ? null : filter === "closed" ? 1 : 0, from?.rank ?? null, from?.created_at ?? null, from?.id ?? null])).rows; + const watches = rows.slice(0, watchPageSize); + const last = rows.length > watchPageSize ? watches[watches.length - 1] : undefined; + return { watches, next_cursor: last ? encodeCursor({ rank: last.rank, created_at: last.cursor_created_at, id: last.id }) : null }; + } + async claim(): Promise { + return (await this.pool.query(`UPDATE bell_monitor_jobs SET lease_until=now()+interval '60 seconds',lease_token=$1,attempts=attempts+1 WHERE id=( + SELECT id FROM bell_monitor_jobs WHERE NOT failed AND next_at<=now() AND (lease_until IS NULL OR lease_until) { + // Running out of work budget is not a failure: keep the saved place, release the attempt, and resume on the next claim. + await this.pool.query("UPDATE bell_monitor_jobs SET lease_until=NULL,lease_token=NULL,last_error=NULL,payload=$3,attempts=$4,next_at=now() WHERE id=$1 AND lease_token=$2", [job.id, job.lease_token, payload, Math.max(0, job.attempts - 1)]); + } + async retry(job: Job, reason: string, terminal: boolean, delay: number) { + await this.pool.query("UPDATE bell_monitor_jobs SET lease_until=NULL,lease_token=NULL,last_error=$3,failed=$4,next_at=now()+$5*interval '1 millisecond' WHERE id=$1 AND lease_token=$2", [job.id, job.lease_token, reason, terminal || job.attempts >= 8, delay]); + } + async schedule() { + await this.transaction(async client => { + const watches = (await client.query<{ id: string }>("UPDATE bell_monitors SET next_scan_at=now()+interval '1 hour' WHERE status='active' AND NOT stopped AND next_scan_at<=now() RETURNING id")).rows; + for (const watch of watches) { + await this.enqueue("sync", `sync:${watch.id}`, watch.id, {}, client); + // Recheck malformed records from the start; other failures retain progress with a renewed resume budget. + await client.query("UPDATE bell_monitor_jobs SET failed=false,attempts=0,next_at=now(),lease_until=NULL,lease_token=NULL,payload=CASE WHEN last_error='invalid_monitor_event' THEN '{}'::jsonb ELSE payload-'resumes' END WHERE id=$1 AND failed AND (lease_until IS NULL OR lease_until console.error("Monitor database connection failed.")); + singleton = new MonitorStore(pool); + } + return singleton; +} diff --git a/server/parallel.ts b/server/parallel.ts index 7451862..49a6b72 100644 --- a/server/parallel.ts +++ b/server/parallel.ts @@ -4,7 +4,7 @@ import { z } from "zod"; const maxCharsTotal = 16_000; const objective = z.string().trim().min(1).max(2_000); const credentialParameter = /(?:^|[_-])auth(?:$|[_-])|authorization|authentication|oauth|token|password|secret|credential|signature|session|api.?key|key-pair-id|^(authcode|code|key|sig|jwt|otp|policy)$/i; -const publicUrl = z.string().trim().max(2_048).transform((value) => value.split("#", 1)[0]).refine((value) => { +export const publicUrl = z.string().trim().max(2_048).transform((value) => value.split("#", 1)[0]).refine((value) => { try { const url = new URL(value); const host = url.hostname.replace(/\.$/, ""); diff --git a/tests/linq.test.ts b/tests/linq.test.ts index cea0849..affba1e 100644 --- a/tests/linq.test.ts +++ b/tests/linq.test.ts @@ -92,7 +92,7 @@ describe("Bell's signed Linq webhook", () => { expect(dispatch).toHaveBeenCalledOnce(); expect(dispatch).toHaveBeenCalledWith(event.data.parts[0].value, expect.objectContaining({ context: [expect.stringContaining("Bell's private Linq")], - auth: expect.objectContaining({ principalId: "linq:sender-handle-id", principalType: "user" }), + auth: expect.objectContaining({ principalId: "linq:sender-handle-id", principalType: "user", attributes: expect.objectContaining({ bell_watch_chat: event.data.chat.id, bell_watch_sender: senderNumber, bell_watch_line: receivingNumber }) }), state: { thread: expect.objectContaining({ isDM: true }) }, })); expect(fetch).toHaveBeenCalledOnce(); diff --git a/tests/monitor-http.test.ts b/tests/monitor-http.test.ts new file mode 100644 index 0000000..b5fac87 --- /dev/null +++ b/tests/monitor-http.test.ts @@ -0,0 +1,119 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { Webhook, WebhookVerificationError } from "standardwebhooks"; +import { receiveWebhook, processMonitors } from "../server/monitors/http"; +import { MonitorService } from "../server/monitors/service"; +import { parallelMonitors, ProviderError, readBounded, sendLinq } from "../server/monitors/provider"; +import { frequencyInput, monitorTimestamp } from "../server/monitors/config"; +import { eventSchema } from "../server/monitors/provider"; +import { watchDestination, watchIdentity } from "../agent/lib/monitors"; +import { MonitorStore, monitorStore } from "../server/monitors/store"; +import { Pool } from "pg"; +import type { ToolContext } from "eve/tools"; + +afterEach(() => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); }); +describe("monitor HTTP boundaries", () => { + it.each(["1h", "6h", "1d", "1w", "30d", "720h", "4w"])("accepts supported frequency %s", value => { + expect(frequencyInput.safeParse(value).success).toBe(true); + }); + it.each(["0h", "5m", "1.5h", "31d", "5w", "721h", "-1h", "daily", ""])("rejects unsupported frequency %s", value => { + expect(frequencyInput.safeParse(value).success).toBe(false); + }); + it("normalizes equivalent timestamps without losing fractional precision", () => { + expect(monitorTimestamp.parse("2026-10-30T05:00:00.123456-07:00")).toBe("2026-10-30T12:00:00.123456Z"); + expect(monitorTimestamp.parse("2026-10-30T12:00:00.000+00:00")).toBe("2026-10-30T12:00:00Z"); + for (const timestamp of ["2026-10-30T12:00:00Z", "2026-10-30T05:00:00-07:00", "2026-10-30T12:00:00+00:00"]) { + for (const event_type of ["completion", "error"]) expect(eventSchema.parse({ event_type, timestamp, error_message: "quota" })).toMatchObject({ timestamp: "2026-10-30T12:00:00Z" }); + } + }); + it("passes selected frequency to create and updates only that field on the existing monitor", async () => { + vi.stubEnv("PARALLEL_API_KEY", "test"); + const fetcher = vi.fn(async () => Response.json({ monitor_id: "monitor-id", status: "active", frequency: "6h" })); + vi.stubGlobal("fetch", fetcher); + await parallelMonitors.create("Public releases", "watch-id", "https://bell.example.com", "6h"); + await parallelMonitors.update("monitor-id", "6h"); + const calls = fetcher.mock.calls as unknown as [string, RequestInit][]; + expect(JSON.parse(calls[0][1].body as string).frequency).toBe("6h"); + expect(calls[1][0]).toBe("https://api.parallel.ai/v1/monitors/monitor-id/update"); + expect(JSON.parse(calls[1][1].body as string)).toEqual({ frequency: "6h" }); + }); + it.each([ + ["monitor.event.detected", { event_group_id: "group-1" }], + ["monitor.execution.completed", { type: "completion", monitor_ts: "completed_2026-10-30T12:00:00Z" }], + ["monitor.execution.failed", { type: "error", error: "Execution failed", id: "error-1", date: "2026-10-30T12:00:00Z" }], + ])("accepts a signed Parallel %s callback", async (type, event) => { + const secret = `whsec_${Buffer.alloc(32, 3).toString("base64")}`; + vi.stubEnv("PARALLEL_WEBHOOK_SECRET", secret); + const pool = new Pool(); + const store = new MonitorStore(pool); + const transaction = vi.spyOn(store, "transaction").mockResolvedValue(undefined); + const service = new MonitorService(store); + const now = new Date(); + const body = JSON.stringify({ type, timestamp: now.toISOString().replace("Z", "+00:00"), data: { monitor_id: "monitor-1", event, metadata: {} } }); + const headers = { "webhook-id": "callback-1", "webhook-timestamp": String(Math.floor(now.getTime() / 1000)), "webhook-signature": new Webhook(secret).sign("callback-1", now, body) }; + try { + expect((await receiveWebhook(new Request("https://bell.example/webhook", { method: "POST", headers, body }), () => service)).status).toBe(204); + expect(transaction).toHaveBeenCalledOnce(); + } finally { await pool.end(); } + }); + it("handles idle database errors without exposing connection details", async () => { + vi.stubEnv("DATABASE_URL", "postgresql://unused"); + const pool = monitorStore().pool; + const log = vi.spyOn(console, "error").mockImplementation(() => {}); + try { + expect(() => pool.emit("error", new Error("sensitive connection details"))).not.toThrow(); + expect(log).toHaveBeenCalledWith("Monitor database connection failed."); + } finally { await pool.end(); } + }); + it("authenticates both scheduler methods and fails closed without a secret", async () => { + const process = vi.fn(async () => ({ processed: 1 })); + const service = () => ({ process }) as unknown as MonitorService; + const request = (method: string, token: string) => new Request("https://bell.example/process", { method, headers: { Authorization: `Bearer ${token}` } }); + expect((await processMonitors(request("GET", "undefined"), service)).status).toBe(401); + vi.stubEnv("CRON_SECRET", "secret"); + expect((await processMonitors(request("POST", "wrong"), service)).status).toBe(401); + for (const method of ["GET", "POST"]) { + const response = await processMonitors(request(method, "secret"), service); + expect(response.status).toBe(200); expect(response.headers.get("cache-control")).toBe("no-store"); + } + expect(process).toHaveBeenCalledTimes(2); + }); + it("does not acknowledge failed persistence, invalid signatures, or oversized callbacks", async () => { + vi.stubEnv("PARALLEL_WEBHOOK_SECRET", "test"); + const ingest = vi.fn(async () => { throw new Error("database unavailable"); }); + const service = () => ({ ingest }) as unknown as MonitorService; + const request = (body = "{}") => new Request("https://bell.example/webhook", { method: "POST", body }); + expect((await receiveWebhook(request(), service)).status).toBe(503); + ingest.mockRejectedValueOnce(new WebhookVerificationError("invalid")); + expect((await receiveWebhook(request(), service)).status).toBe(401); + expect((await receiveWebhook(request("x".repeat(70_000)), service)).status).not.toBe(204); + expect(ingest).toHaveBeenCalledTimes(2); + }); + it("accepts a valid signature among multiple signatures", () => { + const verifier = new Webhook(`whsec_${Buffer.alloc(32, 3).toString("base64")}`); + const now = new Date(), body = '{"test":true}'; + expect(verifier.verify(body, { "webhook-id": "id", "webhook-timestamp": String(Math.floor(now.getTime()/1000)), "webhook-signature": `v1,invalid ${verifier.sign("id", now, body)}` })).toEqual({ test: true }); + }); + it("rejects web identities and requires trusted destination attributes", () => { + const context = (auth: unknown) => ({ session: { auth: { current: auth } } }) as ToolContext; + expect(() => watchIdentity(context({ authenticator: "basic", principalType: "user", principalId: "linq:fake", issuer: "linq" }))).toThrow(); + expect(() => watchDestination(context({ authenticator: "linq-message", principalType: "user", principalId: "linq:real", issuer: "linq", attributes: {} }))).toThrow(); + }); + it("uses the v1 create contract and passes stable Linq idempotency in the message body", async () => { + vi.stubEnv("PARALLEL_API_KEY", "parallel-test"); vi.stubEnv("LINQ_API_KEY", "linq-test"); + const fetcher = vi.fn(async (url: string) => Response.json(url.includes("parallel.ai") ? { monitor_id: "monitor-id", status: "active", frequency: "1d" } : { message: { id: "message-id" } })); + vi.stubGlobal("fetch", fetcher); + await parallelMonitors.create("Public releases", "watch-id", "https://bell.example.com"); + expect(fetcher).toHaveBeenCalledWith("https://api.parallel.ai/v1/monitors", expect.objectContaining({ redirect: "error", headers: expect.objectContaining({ "x-api-key": "parallel-test" }), body: expect.any(String) })); + const init = (fetcher.mock.calls[0] as unknown as [string, RequestInit])[1]; + expect(JSON.parse(init.body as string)).toMatchObject({ type: "event_stream", frequency: "1d", processor: "lite", settings: { include_backfill: false }, metadata: { bell_watch_id: "watch-id" } }); + await sendLinq({ chat: "chat-id", sender: "+12025550123", line: "+12025550100" }, "text", "stable"); + expect(JSON.parse((fetcher.mock.calls[1] as unknown as [string, RequestInit])[1].body as string)).toEqual({ message: { parts: [{ type: "text", value: "text" }], idempotency_key: "stable" } }); + }); + it("bounds response reads and exposes retry delay without leaking error bodies", async () => { + await expect(readBounded(new Response("too long"), 3)).rejects.toThrow("size limit"); + vi.stubEnv("PARALLEL_API_KEY", "test"); + vi.stubGlobal("fetch", vi.fn(async () => new Response("secret provider detail", { status: 429, headers: { "Retry-After": "60" } }))); + await expect(parallelMonitors.list()).rejects.toMatchObject({ status: 429, retryAfter: 60_000 }); + await expect(parallelMonitors.list()).rejects.toBeInstanceOf(ProviderError); + }); +}); diff --git a/tests/monitors.test.ts b/tests/monitors.test.ts new file mode 100644 index 0000000..2cacbdd --- /dev/null +++ b/tests/monitors.test.ts @@ -0,0 +1,538 @@ +import { randomUUID } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import { Pool } from "pg"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { Webhook } from "standardwebhooks"; +import { MonitorService } from "../server/monitors/service"; +import { MonitorStore } from "../server/monitors/store"; +import { type Destination, type Owner } from "../server/monitors/config"; +import { type MonitorProvider, ProviderError } from "../server/monitors/provider"; + +const databaseUrl = process.env.TEST_MONITOR_DATABASE_URL; +if (process.env.CI && !databaseUrl) throw new Error("TEST_MONITOR_DATABASE_URL is required in CI"); +const owner: Owner = { issuer: "linq", id: "linq:sender" }; +const destination: Destination = { chat: "chat-1", sender: "+12025550123", line: "+12025550100" }; +const secret = `whsec_${Buffer.alloc(32, 7).toString("base64")}`; +const detected = { event_type: "event_stream", event_id: "event-1", event_group_id: "group-1", event_date: "2026-10-30", output: { type: "text", content: "A new public release.", basis: [{ citations: [{ url: "https://example.com/release" }] }] } }; + +describe.skipIf(!databaseUrl)("Persistent Parallel watches", () => { + const schema = `monitor_${randomUUID().replaceAll("-", "")}`; + let admin: Pool, pool: Pool, store: MonitorStore, service: MonitorService; + let provider: MonitorProvider; + let send: ReturnType Promise>>; + let remote: { monitor_id: string; status: string; frequency: string; metadata: { bell_watch_id: string } }[]; + beforeAll(async () => { + admin = new Pool({ connectionString: databaseUrl }); + await admin.query(`CREATE SCHEMA ${schema}`); + pool = new Pool({ connectionString: databaseUrl, options: `-c search_path=${schema}`, max: 8 }); + await pool.query(await readFile(new URL("../migrations/004_parallel_monitors.sql", import.meta.url), "utf8")); + await pool.query(await readFile(new URL("../migrations/005_monitor_frequency.sql", import.meta.url), "utf8")); + }); + beforeEach(async () => { + await pool.query("TRUNCATE bell_monitor_jobs, bell_monitor_events, bell_monitor_inbox, bell_monitors CASCADE"); + vi.stubEnv("DATABASE_URL", databaseUrl!); + vi.stubEnv("PARALLEL_API_KEY", "test"); + vi.stubEnv("PARALLEL_WEBHOOK_SECRET", secret); + vi.stubEnv("CRON_SECRET", "worker-secret"); + vi.stubEnv("LINQ_API_KEY", "test"); + vi.stubEnv("BELL_PUBLIC_URL", "https://bell.example.com"); + vi.stubEnv("BELL_LINQ_PHONE_NUMBER", destination.line); + vi.stubEnv("BELL_LINQ_ALLOWED_SENDERS", destination.sender); + remote = []; + provider = { + create: vi.fn(async (_query, id, _origin, frequency = "1d") => { + const result = { monitor_id: `monitor-${id}`, status: "active", frequency, metadata: { bell_watch_id: id } }; + remote.push(result); return result; + }), + list: vi.fn(async () => ({ monitors: remote, next_cursor: null })), + events: vi.fn(async () => ({ events: [detected], next_cursor: null })), + update: vi.fn(async (id, frequency) => { + const monitor = remote.find(m => m.monitor_id === id)!; + monitor.frequency = frequency; + return { ...monitor }; + }), + cancel: vi.fn(async id => { remote.find(m => m.monitor_id === id)!.status = "cancelled"; }), + }; + send = vi.fn(async () => "accepted-message"); + store = new MonitorStore(pool); + service = new MonitorService(store, provider, send); + }); + afterEach(() => vi.unstubAllEnvs()); + afterAll(async () => { await pool?.end(); if (admin) { await admin.query(`DROP SCHEMA ${schema} CASCADE`); await admin.end(); } }); + const create = (operation = "session/call") => service.create(owner, destination, operation, "Watch public product releases"); + const listed = async () => (await service.list(owner)).watches; + async function callback(id: string, webhookId = randomUUID()) { + const body = JSON.stringify({ type: "monitor.event.detected", timestamp: new Date().toISOString(), data: { monitor_id: `monitor-${id}`, event: { event_group_id: "group-1" }, metadata: { bell_watch_id: id } } }); + const timestamp = new Date(); + return service.ingest(body, { "webhook-id": webhookId, "webhook-timestamp": String(Math.floor(timestamp.getTime() / 1000)), "webhook-signature": new Webhook(secret).sign(webhookId, timestamp, body) }); + } + it("upgrades existing daily watches without resetting their state", async () => { + const legacySchema = `${schema}_legacy`; + await admin.query(`CREATE SCHEMA ${legacySchema}`); + const legacy = new Pool({ connectionString: databaseUrl, options: `-c search_path=${legacySchema}` }); + try { + await legacy.query(await readFile(new URL("../migrations/004_parallel_monitors.sql", import.meta.url), "utf8")); + await legacy.query("INSERT INTO bell_monitors(id,issuer,owner_id,operation,query,destination,status) VALUES($1,'linq','linq:sender','call','query',$2,'active')", [randomUUID(), destination]); + const migration = await readFile(new URL("../migrations/005_monitor_frequency.sql", import.meta.url), "utf8"); + await legacy.query(migration); await legacy.query(migration); + expect((await legacy.query("SELECT status,frequency,creation_frequency,pending_frequency FROM bell_monitors")).rows).toEqual([{ status: "active", frequency: "1d", creation_frequency: "1d", pending_frequency: null }]); + } finally { await legacy.end(); await admin.query(`DROP SCHEMA ${legacySchema} CASCADE`); } + }); + it("defaults to daily and preserves an explicit frequency through creation and listing", async () => { + expect(await create()).toMatchObject({ frequency: "1d" }); + const watch = await service.create(owner, destination, "six-hour", "Public releases", "6h"); + expect(watch.frequency).toBe("6h"); + expect(remote.find(m => m.monitor_id === `monitor-${watch.id}`)?.frequency).toBe("6h"); + expect((await listed()).find(w => w.id === watch.id)?.frequency).toBe("6h"); + await expect(service.create(owner, destination, "six-hour", "Public releases", "1d")).rejects.toThrow("different input"); + }); + it("keeps creation replay stable after a confirmed frequency change", async () => { + const watch = await create(); + expect(await service.update(owner, watch.id, "6h")).toMatchObject({ frequency: "1d", pending_frequency: "6h" }); + expect((await listed())[0]).toMatchObject({ frequency: "1d", pending_frequency: "6h" }); + await new MonitorService(new MonitorStore(pool), provider, send).process(); + expect(await create()).toMatchObject({ frequency: "6h", pending_frequency: null }); + expect(provider.create).toHaveBeenCalledTimes(1); + expect(provider.update).toHaveBeenCalledWith(`monitor-${watch.id}`, "6h"); + expect(await service.update(owner, watch.id, "6h")).toMatchObject({ pending_frequency: null }); + await service.process(); + expect(provider.update).toHaveBeenCalledTimes(1); + }); + it("does not overwrite a confirmed cadence with a delayed creation response", async () => { + const watch = await create(); + await service.update(owner, watch.id, "6h"); await service.process(); + await store.bind(watch.id, `monitor-${watch.id}`, false, "1d"); + expect((await store.get(watch.id)).frequency).toBe("6h"); + }); + it("rejects invalid cadence before creating or changing a watch", async () => { + await expect(service.create(owner, destination, "bad", "Public releases", "5m")).rejects.toThrow(); + expect(provider.create).not.toHaveBeenCalled(); + const watch = await create(); + await expect(service.update(owner, watch.id, "31d")).rejects.toThrow(); + expect((await pool.query("SELECT * FROM bell_monitor_jobs WHERE kind='update'")).rowCount).toBe(0); + const prepared = await store.prepare(owner, destination, "not-active", "Public releases"); + await expect(service.update(owner, prepared.id, "6h")).rejects.toThrow("active watch"); + }); + it("scopes updates to their owner and permits only one pending change", async () => { + const watch = await create(); + await expect(service.update({ issuer: "linq", id: "linq:other" }, watch.id, "6h")).rejects.toThrow("Watch not found"); + await Promise.all([service.update(owner, watch.id, "7d"), service.update(owner, watch.id, "1w")]); + expect((await pool.query("SELECT * FROM bell_monitor_jobs WHERE kind='update'")).rowCount).toBe(1); + await expect(service.update(owner, watch.id, "6h")).rejects.toThrow("pending frequency change"); + await service.stop(owner, watch.id); + await service.process(); + expect(provider.update).not.toHaveBeenCalled(); + await expect(service.update(owner, watch.id, "6h")).rejects.toThrow("active watch"); + }); + it("retries a lost update response after restart without claiming premature success", async () => { + const watch = await create(); + const update = provider.update; + provider.update = vi.fn(async (id, frequency) => { await update(id, frequency); throw new Error("response lost"); }); + await service.update(owner, watch.id, "6h"); + await service.process(); + expect((await listed())[0]).toMatchObject({ frequency: "1d", pending_frequency: "6h", problems: [expect.objectContaining({ kind: "update" })] }); + provider.update = update; + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE kind='update'"); + await new MonitorService(new MonitorStore(pool), provider, send).process(); + expect((await listed())[0]).toMatchObject({ frequency: "6h", pending_frequency: null }); + expect(provider.create).toHaveBeenCalledTimes(1); + }); + it("retains a failed update for repair and rejects an unconfirmed response", async () => { + const watch = await create(); + provider.update = vi.fn(async id => ({ ...remote[0], monitor_id: id, frequency: "1d" })); + await service.update(owner, watch.id, "6h"); await service.process(); + expect((await store.get(watch.id)).pending_frequency).toBe("6h"); + provider.update = vi.fn(async () => { throw new ProviderError(403); }); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE kind='update'"); + await service.process(); + expect((await listed())[0]).toMatchObject({ frequency: "1d", pending_frequency: "6h", problems: [expect.objectContaining({ kind: "update", failed: true, error: "provider_http_403" })] }); + await service.stop(owner, watch.id); + expect((await store.get(watch.id)).pending_frequency).toBeNull(); + }); + it("suppresses a claimed update after stop commits", async () => { + const watch = await create(); + await service.update(owner, watch.id, "6h"); + const get = store.get.bind(store); + vi.spyOn(store, "get").mockImplementationOnce(async id => { + const snapshot = await get(id); + await new MonitorStore(pool).stop(owner, id); + return snapshot; + }); + await service.process(); + expect(provider.update).not.toHaveBeenCalled(); + expect((await get(watch.id)).pending_frequency).toBeNull(); + }); + it("cancels after an in-flight update completes", async () => { + const watch = await create(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const update = provider.update; + provider.update = vi.fn(async (id, frequency) => { await gate; return update(id, frequency); }); + await service.update(owner, watch.id, "6h"); + const processing = service.process(); + let stopping: Promise | undefined; + try { + await vi.waitFor(() => expect(provider.update).toHaveBeenCalledOnce()); + stopping = new MonitorStore(pool).stop(owner, watch.id); + await vi.waitFor(async () => expect((await pool.query("SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%owner_id=$3 FOR UPDATE%' ")).rowCount).toBeGreaterThan(0)); + } finally { release(); await Promise.all([processing, stopping]); } + await service.process(); + expect(remote[0]).toMatchObject({ status: "cancelled", frequency: "6h" }); + expect((await store.get(watch.id)).pending_frequency).toBeNull(); + }); + it("processes valid records while surfacing a malformed record as incomplete recovery", async () => { + await create(); + provider.events = vi.fn(async () => ({ events: [detected, { event_type: "completion", timestamp: "invalid" }] })); + await service.process(); + expect(send).toHaveBeenCalledOnce(); + expect((await listed())[0]).toMatchObject({ recovery_gap: true, last_scan_at: null, problems: [expect.objectContaining({ kind: "sync", error: "invalid_monitor_event" })] }); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE kind='sync'"); + await service.process(); + expect(send).toHaveBeenCalledOnce(); + }); + it("deduplicates concurrent creation, scopes ownership, and rejects changed replay input", async () => { + const [a, b] = await Promise.all([create(), create()]); + expect(a.id).toBe(b.id); + expect(provider.create).toHaveBeenCalledTimes(1); + expect(await service.list({ issuer: "linq", id: "linq:other" })).toMatchObject({ watches: [], next_cursor: null }); + await expect(service.stop({ issuer: "linq", id: "linq:other" }, a.id)).rejects.toThrow("Watch not found"); + await expect(service.create(owner, destination, "session/call", "Changed query")).rejects.toThrow("different input"); + }); + it("reconciles an accepted creation with a lost response without another create", async () => { + const original = provider.create; + provider.create = vi.fn(async (query, id, origin) => { await original(query, id, origin); throw new Error("response lost"); }); + const watch = await create(); + expect(watch.status).toBe("uncertain"); + await create(); + await service.process(); + expect(provider.create).toHaveBeenCalledTimes(1); + expect((await listed())[0].status).toBe("active"); + }); + it("delivers a signed event after service replacement without an Eve session, once across workers", async () => { + const watch = await create(); + const webhookId = randomUUID(); + await callback(watch.id, webhookId); + await callback(watch.id, webhookId); + service = new MonitorService(new MonitorStore(pool), provider, send); + await Promise.all([service.process(), service.process()]); + await service.process(); + expect(send).toHaveBeenCalledTimes(1); + expect(send.mock.calls[0][0]).toEqual(destination); + expect(send.mock.calls[0][1]).toContain("https://example.com/release"); + expect((await listed())[0].updates).toHaveLength(1); + }); + it("reuses frozen delivery and key after an accepted send loses its response", async () => { + const watch = await create(); await callback(watch.id); + send.mockRejectedValueOnce(new Error("response lost")); + await service.process(); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE kind='send'"); + await service.process(); + expect(send).toHaveBeenCalledTimes(2); + expect(send.mock.calls[0]).toEqual(send.mock.calls[1]); + }); + it("stops queued deliveries and cancels without webhook or messaging configuration", async () => { + const watch = await create(); await callback(watch.id); + vi.stubEnv("LINQ_API_KEY", ""); vi.stubEnv("PARALLEL_WEBHOOK_SECRET", ""); + await service.stop(owner, watch.id); await service.process(); + expect(send).not.toHaveBeenCalled(); + expect(provider.cancel).toHaveBeenCalledWith(`monitor-${watch.id}`); + expect((await listed())[0].status).toBe("cancelled"); + }); + it("suppresses a claimed send when stop commits after the worker reads the watch", async () => { + const watch = await create(); + await pool.query("UPDATE bell_monitors SET next_scan_at=now()+interval '1 hour'"); + await pool.query("DELETE FROM bell_monitor_jobs"); + await store.enqueue("send", "queued-send", watch.id, { text: "update", key: "stable", eventId: "event-1" }); + const get = store.get.bind(store); + vi.spyOn(store, "get").mockImplementationOnce(async id => { + const snapshot = await get(id); + await new MonitorService(new MonitorStore(pool), provider, send).stop(owner, id); + return snapshot; + }); + await service.process(); + expect(send).not.toHaveBeenCalled(); + expect((await get(watch.id)).stopped).toBe(true); + }); + it("waits for an in-flight send before confirming stop", async () => { + const watch = await create(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + send.mockImplementationOnce(async () => { await gate; return "accepted-message"; }); + const processing = service.process(); + let stopping: Promise | undefined; + try { + await vi.waitFor(() => expect(send).toHaveBeenCalledOnce()); + stopping = new MonitorStore(pool).stop(owner, watch.id); + await vi.waitFor(async () => { + const waiting = await pool.query("SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%owner_id=$3 FOR UPDATE%'"); + expect(waiting.rowCount).toBeGreaterThan(0); + }); + expect((await store.get(watch.id)).stopped).toBe(false); + } finally { + release(); + await Promise.all([processing, stopping]); + } + expect((await store.get(watch.id)).stopped).toBe(true); + await service.process(); + expect(send).toHaveBeenCalledOnce(); + }); + it.each([ + ["monitor.execution.completed", { type: "completion", monitor_ts: "completed_2026-10-30T12:00:00Z" }, "completed"], + ["monitor.execution.failed", { type: "error", error: "Execution failed", id: "error-1", date: "2026-10-30T12:00:00Z" }, "failed"], + ])("stores and processes a signed %s without sending an alert", async (type, event, health) => { + const watch = await create(); + provider.events = vi.fn(async () => ({ events: [] })); + const now = new Date(), id = randomUUID(); + const body = JSON.stringify({ type, timestamp: now.toISOString().replace("Z", "+00:00"), data: { monitor_id: `monitor-${watch.id}`, event } }); + await service.ingest(body, { "webhook-id": id, "webhook-timestamp": String(Math.floor(now.getTime() / 1000)), "webhook-signature": new Webhook(secret).sign(id, now, body) }); + expect((await pool.query("SELECT id FROM bell_monitor_inbox WHERE id=$1", [id])).rowCount).toBe(1); + await service.process(); + expect((await store.get(watch.id)).health).toBe(health); + expect(send).not.toHaveBeenCalled(); + }); + it("suppresses delivery when the sender is removed from the allowlist", async () => { + const watch = await create(); await callback(watch.id); + vi.stubEnv("BELL_LINQ_ALLOWED_SENDERS", ""); + await service.process(); + expect(send).not.toHaveBeenCalled(); + }); + it("rejects forged and stale signatures without durable ingestion", async () => { + await expect(service.ingest("{}", { "webhook-id": "fake", "webhook-timestamp": "0", "webhook-signature": "v1,fake" })).rejects.toThrow(); + const date = new Date(Date.now() - 600_000), body = "{}"; + await expect(service.ingest(body, { "webhook-id": "old", "webhook-timestamp": String(Math.floor(date.getTime()/1000)), "webhook-signature": new Webhook(secret).sign("old", date, body) })).rejects.toThrow(); + expect((await pool.query("SELECT * FROM bell_monitor_inbox")).rowCount).toBe(0); + }); + it("recovers valid events beside offset timestamps and preserves equivalent scan anchors", async () => { + const watch = await create(); + await pool.query("UPDATE bell_monitors SET scan_anchor='completion:2026-10-30T12:00:00Z' WHERE id=$1", [watch.id]); + provider.events = vi.fn(async () => ({ events: [ + { event_type: "completion", timestamp: "2026-10-30T05:00:00-07:00" }, + { event_type: "error", timestamp: "2026-10-29T12:00:00+00:00", error_message: "quota" }, detected, + ] })); + await service.process(); + expect(send).toHaveBeenCalledOnce(); + expect((await listed())[0]).toMatchObject({ health: "completed", recovery_gap: false, problems: [] }); + }); + it("records completion and errors without event IDs and does not send health messages", async () => { + await create(); + provider.events = vi.fn(async () => ({ events: [{ event_type: "completion", timestamp: "2026-10-30T12:00:00Z" }, { event_type: "error", timestamp: "2026-10-29T12:00:00Z", error_message: "quota" }], next_cursor: null })); + await service.process(); + expect(send).not.toHaveBeenCalled(); + const [watch] = await listed(); + expect(watch.health).toBe("completed"); + }); + it("never redispatches creation after a crash at the dispatch boundary", async () => { + const watch = await store.prepare(owner, destination, "crashed", "Watch public releases"); + expect(await store.dispatch(watch.id)).toBe(true); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now()"); + await service.create(owner, destination, "crashed", "Watch public releases"); + await service.process(); + expect(provider.create).not.toHaveBeenCalled(); + expect((await store.get(watch.id)).remote_id).toBeNull(); + }); + it("cancels an uncertain remote creation when stop arrives before reconciliation", async () => { + const original = provider.create; + provider.create = vi.fn(async (query, id, origin) => { await original(query, id, origin); throw new Error("lost"); }); + const watch = await create(); + await service.stop(owner, watch.id); + await service.process(); + expect(remote[0].status).toBe("cancelled"); + expect((await listed())[0].status).toBe("cancelled"); + expect(send).not.toHaveBeenCalled(); + }); + it("retains an early callback until remote identity can be verified", async () => { + const original = provider.create; + provider.create = vi.fn(async (query, id, origin) => { + await callback(id); + await service.process(); + return original(query, id, origin); + }); + await create(); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now()"); + await service.process(); + expect(send).toHaveBeenCalledTimes(1); + }); + it("resumes an expired delivery lease with the same frozen payload", async () => { + const watch = await create(); + await callback(watch.id); + send.mockRejectedValueOnce(new Error("interrupted")); + await service.process(); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now(),lease_until=now()-interval '1 minute',lease_token=$1 WHERE kind='send'", [randomUUID()]); + await new MonitorService(new MonitorStore(pool), provider, send).process(); + expect(send.mock.calls[0]).toEqual(send.mock.calls[1]); + }); + it("records local suppression when the provider key is missing", async () => { + const watch = await create(); + vi.stubEnv("PARALLEL_API_KEY", ""); + provider.cancel = vi.fn(async () => { throw new Error("missing key"); }); + expect((await service.stop(owner, watch.id)).status).toBe("cancel_pending"); + await service.process(); + expect((await store.get(watch.id)).stopped).toBe(true); + expect((await listed())[0].problems).not.toEqual([]); + expect(send).not.toHaveBeenCalled(); + }); + it("surfaces lost history and handles overlapping pages without duplicate sends", async () => { + const watch = await create(); + await service.process(); + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + const newer = { ...detected, event_id: "event-2", event_group_id: "group-2" }; + provider.events = vi.fn(async (_id, _group, cursor) => cursor ? { events: [newer], next_cursor: null } : { events: [newer], next_cursor: "page-2" }); + await service.process(); + expect(send).toHaveBeenCalledTimes(2); + expect((await listed())[0].recovery_gap).toBe(true); + }); + it("resumes a recovery scan stopped by the work budget without refetching pages or resending events", async () => { + const watch = await create(); + await service.process(); + expect(send).toHaveBeenCalledTimes(1); + const page = (id: number, next: string | null) => ({ events: [{ ...detected, event_id: `event-${id}`, event_group_id: `group-${id}` }], next_cursor: next }); + const pages: Record = { + start: page(2, "cursor-2"), "cursor-2": page(3, "cursor-3"), "cursor-3": page(4, "cursor-4"), + // The oldest page overlaps the previous scan's anchor, so coverage is complete and event-1 must not be resent. + "cursor-4": { events: [{ ...detected, event_id: "event-5", event_group_id: "group-5" }, detected], next_cursor: null }, + }; + const fetched: string[] = []; + provider.events = vi.fn(async (_id, _group, cursor) => { + fetched.push(cursor ?? "start"); + await new Promise(resolve => setTimeout(resolve, 80)); + return pages[cursor ?? "start"]; + }); + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + // A work budget smaller than one page leaves each run able to fetch a single page, as a slow recovery does against the real 20 seconds. + const slow = new MonitorService(store, provider, send, 50); + const saved: unknown[] = []; + for (let run = 0; run < 10; run++) { + await slow.process(); + const job = (await pool.query<{ attempts: number; failed: boolean; last_error: string | null; payload: { cursor?: string } }>("SELECT attempts,failed,last_error,payload FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rows[0]; + if (!job) break; + expect(job).toMatchObject({ attempts: 0, failed: false, last_error: null }); + saved.push(job.payload.cursor); + } + expect(saved).toEqual(["cursor-2", "cursor-3", "cursor-4"]); + expect(fetched).toEqual(["start", "cursor-2", "cursor-3", "cursor-4"]); + await service.process(); + const [listedWatch] = await listed(); + expect(listedWatch).toMatchObject({ recovery_gap: false, problems: [] }); + expect(listedWatch.last_scan_at).not.toBeNull(); + expect(send).toHaveBeenCalledTimes(5); + expect(new Set(send.mock.calls.map(call => call[2])).size).toBe(5); + }); + it("keeps a budget-limited recovery scan alive past the eight-attempt limit", async () => { + const watch = await create(); + await service.process(); + provider.events = vi.fn(async (_id, _group, cursor) => { + const read = cursor ? Number(cursor.slice("cursor-".length)) : 1; + await new Promise(resolve => setTimeout(resolve, 80)); + if (read > 11) return { events: [detected], next_cursor: null }; + return { events: [{ ...detected, event_id: `event-${read}`, event_group_id: `group-${read}` }], next_cursor: `cursor-${read + 1}` }; + }); + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + // Twelve pages at one page per budget-limited run take more runs than the eight attempts that used to mark the job failed. + const slow = new MonitorService(store, provider, send, 50); + let runs = 0; + for (; runs < 20; runs++) { + await slow.process(); + const job = (await pool.query<{ attempts: number; failed: boolean }>("SELECT attempts,failed FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rows[0]; + if (!job) break; + expect(job).toMatchObject({ attempts: 0, failed: false }); + } + expect(runs).toBeGreaterThan(8); + await service.process(); + expect((await listed())[0]).toMatchObject({ recovery_gap: false, problems: [] }); + expect(new Set(send.mock.calls.map(call => call[2])).size).toBe(send.mock.calls.length); + }); + it("restarts a recovery scan that already exhausted its attempts on the next scheduled check", async () => { + const watch = await create(); + await service.process(); + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + provider.events = vi.fn(async () => { throw new Error("recovery unavailable"); }); + for (let attempt = 0; attempt < 8; attempt++) { + await service.process(); + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE id=$1", [`sync:${watch.id}`]); + } + expect((await pool.query<{ failed: boolean }>("SELECT failed FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rows[0]).toMatchObject({ failed: true }); + provider.events = vi.fn(async () => ({ events: [detected], next_cursor: null })); + await service.process(); + expect(provider.events).not.toHaveBeenCalled(); + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + await service.process(); + expect(provider.events).toHaveBeenCalled(); + expect((await pool.query("SELECT 1 FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rowCount).toBe(0); + expect((await listed())[0].last_scan_at).not.toBeNull(); + }); + it.each(["malformed record", "resume limit"])("recovers after a saved scan fails on its %s", async failure => { + const watch = await create(); + let clock = Date.now(), repaired = false; + vi.spyOn(Date, "now").mockImplementation(() => clock); + const lastPage = failure === "malformed record" ? 3 : 63; + provider.events = vi.fn(async (_id, _group, cursor) => { + clock += repaired && failure === "malformed record" ? 1 : 7_000; + const page = Number(cursor ?? 0); + const events: unknown[] = [{ event_type: "completion", timestamp: new Date(Date.UTC(2026, 9, 30) - page * 3_600_000).toISOString() }]; + if (failure === "malformed record" && !page) events.push(repaired ? detected : { event_type: "completion", timestamp: "invalid" }); + if (failure === "resume limit" && page === lastPage) events.push(detected); + return { events, next_cursor: page < lastPage ? String(page + 1) : null }; + }); + let failed = false; + for (let run = 0; run < 28; run++) { + await service.process(); + failed = (await pool.query("SELECT failed FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rows[0]?.failed === true; + if (failed) break; + await pool.query("UPDATE bell_monitor_jobs SET next_at=now() WHERE id=$1", [`sync:${watch.id}`]); + } + expect(failed).toBe(true); + repaired = true; + await pool.query("UPDATE bell_monitors SET next_scan_at=now() WHERE id=$1", [watch.id]); + for (let run = 0; run < 3; run++) await new MonitorService(new MonitorStore(pool), provider, send).process(); + expect((await pool.query("SELECT 1 FROM bell_monitor_jobs WHERE id=$1", [`sync:${watch.id}`])).rowCount).toBe(0); + expect((await store.get(watch.id)).last_scan_at).not.toBeNull(); + expect(send).toHaveBeenCalledOnce(); + }); + it("keeps an active watch reachable behind newer cancelled watches and pages the rest", async () => { + const active = await create(); + for (let index = 0; index < 50; index++) { + await pool.query(`INSERT INTO bell_monitors(id,issuer,owner_id,operation,query,destination,status,creation_frequency,frequency,created_at) + VALUES($1,$2,$3,$4,'Cancelled watch',$5,'cancelled','1d','1d',now()+($6||' seconds')::interval)`, [randomUUID(), owner.issuer, owner.id, `cancelled-${index}`, destination, index + 1]); + } + const first = await service.list(owner); + expect(first.watches).toHaveLength(50); + expect(first.watches[0].id).toBe(active.id); + expect(first.next_cursor).toEqual(expect.any(String)); + const second = await service.list(owner, { cursor: first.next_cursor! }); + expect(second.watches).toHaveLength(1); + expect(second.next_cursor).toBeNull(); + expect(new Set([...first.watches, ...second.watches].map(w => w.id)).size).toBe(51); + expect(await service.list(owner, { status: "open" })).toMatchObject({ watches: [expect.objectContaining({ id: active.id })], next_cursor: null }); + const closed = await service.list(owner, { status: "closed" }); + expect(closed.watches).toHaveLength(50); + expect(closed.watches.every(w => w.status === "cancelled")).toBe(true); + expect(closed.next_cursor).toBeNull(); + await expect(service.list(owner, { cursor: "not-a-cursor" })).rejects.toThrow("page cursor"); + await expect(service.list(owner, { status: "retired" })).rejects.toThrow(); + }); + it.each(["2026-10-31T12:00:00.123000Z", "2026-10-31T12:00:00.123456Z"])("pages every watch sharing timestamp %s", async timestamp => { + await pool.query(`INSERT INTO bell_monitors(id,issuer,owner_id,operation,query,destination,status,created_at) + SELECT gen_random_uuid(),$1,$2,'page-'||i,'Public releases',$3,'active',$4::timestamptz FROM generate_series(1,51) i`, [owner.issuer, owner.id, destination, timestamp]); + const first = await service.list(owner); + expect(first.watches).toHaveLength(50); + expect(first.next_cursor).not.toBeNull(); + const second = await service.list(owner, { cursor: first.next_cursor! }); + expect(second.watches).toHaveLength(1); + expect(second.next_cursor).toBeNull(); + expect(new Set([...first.watches, ...second.watches].map(w => w.id)).size).toBe(51); + }); + it("keeps valid citations when another URL is unsafe", async () => { + await create(); + provider.events = vi.fn(async () => ({ events: [{ ...detected, output: { ...detected.output, basis: [{ citations: [{ url: "http://localhost/private" }, { url: "https://example.com/release" }] }] } }], next_cursor: null })); + await service.process(); + expect(send.mock.calls[0][1]).toContain("https://example.com/release"); + expect(send.mock.calls[0][1]).not.toContain("localhost"); + }); + it("does not adopt an unrelated account callback", async () => { + await callback(randomUUID()); + await service.process(); + expect(provider.events).not.toHaveBeenCalled(); + expect(send).not.toHaveBeenCalled(); + }); + +});