Add push-based agent automations

.env.example

@@ -17,6 +17,12 @@ BLOB_STORE_ID=
 BLOB_READ_WRITE_TOKEN=
 # Optional Vercel Connect configuration.
 GOOGLE_CONNECTOR_UID=
+# Gmail event automations use an authenticated Google Cloud Pub/Sub push
+# subscription. The topic must belong to the OAuth connector's Google Cloud
+# project and grant Gmail's publisher service account permission to publish.
+GMAIL_PUBSUB_TOPIC=
+GMAIL_PUBSUB_AUDIENCE=
+GMAIL_PUBSUB_SERVICE_ACCOUNT=
 # Linq delivery requires the connector. The phone number only enables the
 # optional click-to-message shortcut in the workspace.
 LINQ_CONNECTOR=

.gitignore

@@ -12,3 +12,4 @@ next-env.d.ts
 dist
 .DS_Store
 *.tsbuildinfo
+/.swc

README.md

@@ -141,6 +141,38 @@ access deliberately uses `gmail.modify`, not the permanent-delete
 4. Set `GOOGLE_CONNECTOR_UID` to the returned UID and redeploy. The default is
    `google/open-instinct`.
 
+### Gmail push automations
+
+Gmail-triggered automations use Gmail `users.watch` and an authenticated Google
+Cloud Pub/Sub push subscription. They do not poll the inbox. Timer automations
+use Vercel Workflow durable sleeps and likewise do not need a cron dispatcher.
+
+1. In the same Google Cloud project as the OAuth credentials, create a Pub/Sub
+   topic and grant `gmail-api-push@system.gserviceaccount.com` the Pub/Sub
+   Publisher role on that topic.
+2. Create a dedicated service account for Pub/Sub push authentication. Create a
+   push subscription whose endpoint is
+   `https://<deployment>/api/automations/gmail`, enable OIDC authentication with
+   that service account, and set the token audience to that exact endpoint URL.
+3. Configure and redeploy:
+
+   ```bash
+   GMAIL_PUBSUB_TOPIC=projects/<google-cloud-project>/topics/<topic>
+   GMAIL_PUBSUB_AUDIENCE=https://<deployment>/api/automations/gmail
+   GMAIL_PUBSUB_SERVICE_ACCOUNT=<push-service-account>@<google-cloud-project>.iam.gserviceaccount.com
+   ```
+
+OpenInstinct creates a watch only after a signed-in user creates a Gmail
+automation, stores Gmail's history cursor, renews the watch before expiration,
+and deduplicates each matching message before running the saved task. Gmail
+push configuration is intentionally explicit: if any value is absent, Gmail
+automations fail at creation instead of silently falling back to polling.
+Automation executions use a fresh, replay-stable Eve session. The session stays
+resumable until its run binding and result are durable, then OpenInstinct retires
+it. If a task requests a later approval, question response, or OAuth sign-in,
+the run records a visible failure and retires the pending session instead of
+duplicating the request.
+
 Gotchas:
 
 - Attach the connector separately to every Vercel environment that should use

agent/channels/eve.ts

@@ -1,13 +1,21 @@
 import { eveChannel } from "eve/channels/eve";
 import { ForbiddenError, UnauthenticatedError } from "eve/channels/auth";
 import { z } from "zod";
+import {
+  readAutomationById,
+  readAutomationRunById,
+} from "@/db/services/automations";
 import { isSessionOwned } from "@/db/services/sessions";
 import { accessScopeForUser, type AccessScope } from "@/lib/access-scope";
+import { verifyAutomationRequest } from "@/lib/automation-auth";
 import { getAuthSession } from "@/auth/session";
 
 export default eveChannel({
   auth: [
     async (request) => {
+      const automationIdentity = await automationIdentityFromRequest(request);
+      if (automationIdentity) return automationIdentity;
+
       const identity = await requestIdentityFromRequest(request);
       if (!identity) {
         throw new UnauthenticatedError({
@@ -32,6 +40,37 @@ export default eveChannel({
   ],
 });
 
+async function automationIdentityFromRequest(request: Request) {
+  const signed = await verifyAutomationRequest(request.headers, "execute");
+  if (!signed?.runId) return undefined;
+  const [automation, run] = await Promise.all([
+    readAutomationById(signed.automationId),
+    readAutomationRunById(signed.runId),
+  ]);
+  const requestedSessionId = sessionIdFromPath(new URL(request.url).pathname);
+  if (
+    automation?.status !== "active" ||
+    automation.revision !== signed.revision ||
+    run?.automationId !== automation.id ||
+    run.revision !== signed.revision ||
+    run.status !== "running" ||
+    (requestedSessionId !== undefined &&
+      requestedSessionId !== run.eveSessionId)
+  ) {
+    throw new ForbiddenError({ message: "Automation is no longer active." });
+  }
+  return {
+    attributes: {
+      automationId: automation.id,
+      phoneNumber: automation.phoneNumber,
+      workspaceId: automation.workspaceId,
+    },
+    authenticator: "automation",
+    principalId: automation.createdByUserId,
+    principalType: "user" as const,
+  };
+}
+
 function sessionIdFromPath(pathname: string) {
   const match = /^\/eve\/v1\/session\/([^/]+)/.exec(pathname);
   if (!match?.[1]) return undefined;

agent/instructions.md

@@ -39,6 +39,7 @@ The main conversation is the control plane. Coordinate the user's work there, de
 - Perform public research, source discovery, comparisons, and current-information lookups directly with `web_search`. Never delegate a search-only task or use a browser to visit a search engine or browse search-result pages. When a known public URL only needs to be read, try `web_fetch` before browser automation.
 - Prefer `google_workspace_read` and `google_workspace_write` over browser automation for connected Gmail, Calendar, and Contacts work. Never ask for Google tokens or credentials in chat. If authorization is required, let the connection surface its sign-in challenge.
 - Use exact Gmail message IDs for reversible inbox updates. Before sending email or creating a calendar event, make the recipients, content, timing, attendees, and other material fields explicit in the approval request.
+- Use `manage_automations` when the user asks for a later text, recurring task, or Gmail-based alert. Resolve relative dates into an explicit trigger using the user's timezone, save the complete task the future run must perform, and say it is scheduled only after the tool confirms it is armed. Use a Gmail trigger only when the requested event can be expressed by sender, thread, or subject; do not simulate unsupported event sources with polling.
 - Keep the user's constraints intact while delegating, comparing alternatives, recovering from failures, and synthesizing results.
 - When the conversation reveals a useful next action, offer that exact action with the details already established: book the 7:15 showtime, buy the selected groceries, or submit the prepared form. Offer execution, not a generic "anything else?" or instructions for the user to do it themselves.
 - If the user's intent is already clear and the action is authorized, act instead of asking whether to act. Do not add an offer to greetings, simple factual answers, or work you already completed.

agent/tools/manage_automations.ts

@@ -0,0 +1,132 @@
+import { defineTool } from "eve/tools";
+import { z } from "zod";
+import {
+  createAutomation,
+  listAutomations,
+  setAutomationStatus,
+} from "@/db/services/automations";
+import { env } from "@/env";
+import { withGoogleAuth } from "@/agent/lib/google-workspace/client";
+import { scopeFromPrincipal } from "@/lib/access-scope";
+import { createAutomationRequestHeaders } from "@/lib/automation-auth";
+import { internalApplicationOrigin } from "@/lib/application-origin";
+import { assertTimezone, automationTriggerSchema } from "@/lib/automation";
+import { isE164PhoneNumber } from "@/auth/phone-number";
+
+const inputSchema = z.discriminatedUnion("action", [
+  z.object({
+    action: z.literal("create"),
+    task: z.string().min(1).max(10_000),
+    timezone: z.string().min(1).default("America/New_York"),
+    title: z.string().min(1).max(200),
+    trigger: automationTriggerSchema,
+  }),
+  z.object({ action: z.literal("list") }),
+  z.object({
+    action: z.enum(["pause", "resume", "delete"]),
+    automationId: z.string().min(1),
+  }),
+]);
+
+export default defineTool({
+  approval: ({ toolInput }) =>
+    toolInput?.action === "delete" ? "user-approval" : "not-applicable",
+  description:
+    "Create and manage durable push automations. Timers sleep until an exact time without polling. Recurring and interval triggers schedule their next run after each delivery. Gmail triggers use authenticated Gmail push notifications and can match a sender, thread, or subject text. Every trigger runs the saved task with fresh data and texts the result to the authenticated user. Use list before changing an automation when its id is unknown. Deletion requires approval.",
+  inputSchema,
+  async execute(input, ctx) {
+    const principal = ctx.session.auth.current ?? ctx.session.auth.initiator;
+    if (principal?.principalType !== "user") {
+      throw new Error("Automations require an authenticated user.");
+    }
+    if (principal.authenticator === "automation" && input.action !== "list") {
+      throw new Error(
+        "Automation runs cannot change the automation control plane."
+      );
+    }
+    const scope = scopeFromPrincipal(principal);
+
+    if (input.action === "list") {
+      return { automations: await listAutomations(scope) };
+    }
+    if (input.action !== "create") {
+      const status =
+        input.action === "pause"
+          ? "paused"
+          : input.action === "resume"
+            ? "active"
+            : "deleted";
+      const automation = await setAutomationStatus(
+        scope,
+        input.automationId,
+        status
+      );
+      if (!automation) throw new Error("Automation not found.");
+      const armed =
+        status === "active" ? await armAutomation(automation) : undefined;
+      return { armed, automation };
+    }
+
+    assertTimezone(input.timezone);
+    if (
+      input.trigger.kind === "gmail" &&
+      !input.trigger.fromAddress &&
+      !input.trigger.subjectContains &&
+      !input.trigger.threadId
+    ) {
+      throw new Error("A Gmail automation needs at least one message filter.");
+    }
+    if (input.trigger.kind === "gmail") {
+      if (
+        !env.GMAIL_PUBSUB_AUDIENCE ||
+        !env.GMAIL_PUBSUB_SERVICE_ACCOUNT ||
+        !env.GMAIL_PUBSUB_TOPIC
+      ) {
+        throw new Error(
+          "Gmail push automations are not configured on this deployment."
+        );
+      }
+      await withGoogleAuth(ctx, async () => undefined);
+    }
+    const phoneNumber = z
+      .string()
+      .refine(isE164PhoneNumber)
+      .parse(principal.attributes.phoneNumber);
+    const automation = await createAutomation(scope, {
+      idempotencyKey: `${ctx.session.id}:${ctx.callId}`,
+      phoneNumber,
+      sessionId: ctx.session.id,
+      task: input.task,
+      timezone: input.timezone,
+      title: input.title,
+      trigger: input.trigger,
+    });
+    return { armed: await armAutomation(automation), automation };
+  },
+});
+
+async function armAutomation(automation: {
+  readonly id: string;
+  readonly revision: number;
+}) {
+  const headers = await createAutomationRequestHeaders({
+    automationId: automation.id,
+    purpose: "arm",
+    revision: automation.revision,
+  });
+  const response = await fetch(
+    `${internalApplicationOrigin()}/api/automations/arm`,
+    {
+      headers,
+      method: "POST",
+      redirect: "error",
+    }
+  );
+  const body: unknown = await response.json().catch(() => undefined);
+  if (!response.ok) {
+    throw new Error(
+      `Automation was saved but could not be armed (HTTP ${String(response.status)}): ${JSON.stringify(body)}`
+    );
+  }
+  return body;
+}

db/migrations/0006_chilly_the_leader.sql

@@ -0,0 +1,59 @@
+CREATE TABLE "automation_runs" (
+	"id" text PRIMARY KEY NOT NULL,
+	"automation_id" text NOT NULL,
+	"revision" integer NOT NULL,
+	"trigger_key" text NOT NULL,
+	"status" text DEFAULT 'running' NOT NULL,
+	"result" text,
+	"error" text,
+	"started_at" text NOT NULL,
+	"completed_at" text,
+	CONSTRAINT "automation_runs_status_check" CHECK ("automation_runs"."status" IN ('running', 'completed', 'failed', 'suppressed'))
+);
+--> statement-breakpoint
+CREATE TABLE "automations" (
+	"id" text PRIMARY KEY NOT NULL,
+	"workspace_id" text NOT NULL,
+	"created_by_user_id" text NOT NULL,
+	"session_id" text NOT NULL,
+	"phone_number" text NOT NULL,
+	"title" text NOT NULL,
+	"task" text NOT NULL,
+	"trigger" text NOT NULL,
+	"timezone" text NOT NULL,
+	"status" text DEFAULT 'active' NOT NULL,
+	"revision" integer DEFAULT 1 NOT NULL,
+	"next_run_at" text,
+	"last_run_at" text,
+	"idempotency_key" text NOT NULL,
+	"created_at" text NOT NULL,
+	"updated_at" text NOT NULL,
+	CONSTRAINT "automations_status_check" CHECK ("automations"."status" IN ('active', 'paused', 'completed', 'deleted')),
+	CONSTRAINT "automations_revision_check" CHECK ("automations"."revision" > 0)
+);
+--> statement-breakpoint
+CREATE TABLE "gmail_watches" (
+	"workspace_id" text NOT NULL,
+	"user_id" text NOT NULL,
+	"email_address" text,
+	"history_id" text,
+	"expiration_at" text,
+	"generation" integer DEFAULT 1 NOT NULL,
+	"status" text DEFAULT 'arming' NOT NULL,
+	"workflow_run_id" text,
+	"created_at" text NOT NULL,
+	"updated_at" text NOT NULL,
+	CONSTRAINT "gmail_watches_pkey" PRIMARY KEY("workspace_id","user_id"),
+	CONSTRAINT "gmail_watches_status_check" CHECK ("gmail_watches"."status" IN ('arming', 'active', 'paused', 'failed')),
+	CONSTRAINT "gmail_watches_generation_check" CHECK ("gmail_watches"."generation" > 0)
+);
+--> statement-breakpoint
+ALTER TABLE "automation_runs" ADD CONSTRAINT "automation_runs_automation_id_fkey" FOREIGN KEY ("automation_id") REFERENCES "public"."automations"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "automations" ADD CONSTRAINT "automations_membership_fkey" FOREIGN KEY ("workspace_id","created_by_user_id") REFERENCES "public"."workspace_memberships"("workspace_id","user_id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "automations" ADD CONSTRAINT "automations_session_id_fkey" FOREIGN KEY ("session_id") REFERENCES "public"."agent_sessions"("session_id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "gmail_watches" ADD CONSTRAINT "gmail_watches_membership_fkey" FOREIGN KEY ("workspace_id","user_id") REFERENCES "public"."workspace_memberships"("workspace_id","user_id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+CREATE UNIQUE INDEX "automation_runs_trigger_uidx" ON "automation_runs" USING btree ("automation_id","trigger_key");--> statement-breakpoint
+CREATE INDEX "automation_runs_automation_started_idx" ON "automation_runs" USING btree ("automation_id","started_at" DESC NULLS FIRST);--> statement-breakpoint
+CREATE UNIQUE INDEX "automations_workspace_idempotency_uidx" ON "automations" USING btree ("workspace_id","idempotency_key");--> statement-breakpoint
+CREATE INDEX "automations_workspace_status_idx" ON "automations" USING btree ("workspace_id","status","next_run_at");--> statement-breakpoint
+CREATE UNIQUE INDEX "gmail_watches_email_uidx" ON "gmail_watches" USING btree ("email_address");

db/migrations/0007_freezing_silver_sable.sql

@@ -0,0 +1 @@
+ALTER TABLE "automation_runs" ADD COLUMN "eve_session_id" text;