Add persistent Parallel watches

.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=

.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

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) |

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. "

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.

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<T>(operation: () => Promise<T>): Promise<T> {
+  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.");
+  }
+}

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),
+});

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),
+});

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),
+});

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),
+});

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); }

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); }