Phase 2: trust that grows (plans, learned rules, undo window)

README.md

@@ -30,6 +30,10 @@ Agent products that act in your tools share three problems: approvals that live
 - Automations triggered by a cron schedule, new Gmail messages matching a search, or a secret webhook URL
 - MCP connections over Streamable HTTP with OAuth 2.1 + PKCE, or bearer tokens; presets for GitHub, Linear, Notion, Atlassian, Sentry
 - Approval inbox, editable approvals, deny-with-reason, cancel now / after this step
+- Approve a plan, not every step: the agent proposes the exact actions it will take, you untick what you don't want, and the kept actions run without asking again (even after the task read outside content, because you saw each one)
+- Several pending actions from one step can be ticked and signed off together
+- Approvals that learn: after you approve the same kind of action unchanged a few times, Synapse offers a visible, removable rule ("emails where every address is @acme.com"). Tasks that read outside content still ask
+- Undo window: approved sends and deletes wait a few seconds (default 10, configurable) with an Undo button
 - Knowledge base: upload docs, hybrid full-text + vector search (pgvector, reciprocal rank fusion)
 - Memory you can read and edit; the agent can add notes, but not from a tainted run without asking
 - Web search (Tavily, Brave or Exa) and SSRF-guarded web fetch

apps/server/src/app.ts

@@ -8,6 +8,7 @@ import { clientMetadataDocument, connections, oauthCallback } from "./routes/con
 import { googleCallback, integrations } from "./routes/integrations";
 import { models } from "./routes/models";
 import { approvals, runs } from "./routes/runs";
+import { suggestions } from "./routes/suggestions";
 import type { Services } from "./services";
 
 export function createApp(services: Services) {
@@ -65,6 +66,7 @@ export function createApp(services: Services) {
   api.route("/connections", connections);
   api.route("/integrations/google/callback", googleCallback);
   api.route("/integrations", integrations);
+  api.route("/policy/suggestions", suggestions);
   api.route("/policy", policy);
   api.route("/memories", memories);
   api.route("/documents", documents);

apps/server/src/prompt.ts

@@ -9,6 +9,7 @@ How you work:
 - Ask a short clarifying question only when a wrong guess would be costly (wrong recipient, wrong repository, irreversible change). Otherwise make a reasonable assumption and say what you assumed.
 - When you draft something that will be sent or published (an email, a message, an issue), put the complete final content in the tool call. The user reviews exactly that on an approval card.
 - Some actions need the user's approval. The system asks for it automatically; you do not need to ask in text first. If an action is denied, do not retry it in another form; acknowledge it and continue or ask what they want instead.
+- When a task needs several actions that change things (sending, writing, deleting), first call propose_plan with the exact calls you intend to make, so the user can approve them once. After approval, make exactly the approved calls with exactly the same inputs, and nothing the user removed. For a single action, just make the call.
 - If a tool fails, read the error, fix your input once if it is clearly your mistake, and otherwise tell the user what failed.
 - Keep answers concise and skimmable. Cite sources (page titles and URLs, document names) for facts you looked up.
 

apps/server/src/routes/account.ts

@@ -1,5 +1,5 @@
 import { unsafePatternReason } from "@synapse/core";
-import { schema } from "@synapse/db";
+import { DEFAULT_UNDO_SECONDS, MAX_UNDO_SECONDS, schema } from "@synapse/db";
 import { ingestDocument, searchKnowledge } from "@synapse/tools";
 import { and, asc, desc, eq, gte, sql } from "drizzle-orm";
 import { Hono } from "hono";
@@ -12,6 +12,11 @@ const Matcher = z.discriminatedUnion("op", [
   z.object({ path: z.string().min(1), op: z.literal("in"), value: z.array(z.json()) }),
   z.object({ path: z.string().min(1), op: z.literal("allEndWith"), value: z.array(z.string()) }),
   z.object({ path: z.string().min(1), op: z.literal("matches"), value: z.string().max(500) }),
+  z.object({
+    path: z.string().default("*"),
+    op: z.literal("emailsEndWith"),
+    value: z.array(z.string().min(2)).min(1),
+  }),
 ]);
 const Defaults = z.object({
   read: Decision,
@@ -233,6 +238,7 @@ export const settings = new Hono<AppEnv>()
         providerKeys: z
           .partialRecord(z.enum(PROVIDERS), z.string().min(8).max(400).nullable())
           .optional(),
+        undoSeconds: z.number().int().min(0).max(MAX_UNDO_SECONDS).optional(),
       }),
     );
     const { db } = c.var.services.database;
@@ -250,6 +256,7 @@ export const settings = new Hono<AppEnv>()
       ...(input.instructions !== undefined ? { instructions: input.instructions } : {}),
       ...(input.models ? { models: input.models } : {}),
       ...(input.budgets ? { budgets: input.budgets } : {}),
+      ...(input.undoSeconds !== undefined ? { undoSeconds: input.undoSeconds } : {}),
       providerKeys: keys,
     };
     await db
@@ -273,6 +280,7 @@ async function loadSettings(c: { var: AppEnv["Variables"] }) {
       perDayUsd: config.DAY_MAX_COST_USD,
     },
     providerKeys: Object.fromEntries(PROVIDERS.map((p) => [p, Boolean(s?.providerKeys?.[p])])),
+    undoSeconds: s?.undoSeconds ?? DEFAULT_UNDO_SECONDS,
   };
 }
 

apps/server/src/routes/chats.ts

@@ -73,6 +73,7 @@ export const chats = new Hono<AppEnv>()
           output: r.output,
           decidedBy: r.decidedBy,
           preview: r.preview,
+          executeAfter: r.executeAfter?.toISOString() ?? null,
         }))
       : [];
     const [run] = await db

apps/server/src/routes/runs.ts

@@ -1,11 +1,20 @@
-import { evaluatePolicy } from "@synapse/core";
-import { decideToolCall, PgRunStore, readEvents, requestCancel, schema } from "@synapse/db";
+import { evaluatePolicy, type RunEvent, UNDO_DENIAL } from "@synapse/core";
+import {
+  decideToolCall,
+  decideToolCalls,
+  PgRunStore,
+  readEvents,
+  requestCancel,
+  schema,
+  undoToolCall,
+} from "@synapse/db";
 import type { JSONValue } from "ai";
 import { and, asc, desc, eq } from "drizzle-orm";
 import { Hono } from "hono";
 import { streamSSE } from "hono/streaming";
 import { z } from "zod";
 import { type AppEnv, apiError, body } from "../http";
+import type { Services } from "../services";
 
 const PAUSE_OR_END = new Set(["waiting", "completed", "failed", "cancelled"]);
 const EVENT_PAGE = 1000;
@@ -53,6 +62,7 @@ export const runs = new Hono<AppEnv>()
         output: schema.toolCalls.output,
         decidedBy: schema.toolCalls.decidedBy,
         decidedAt: schema.toolCalls.decidedAt,
+        executeAfter: schema.toolCalls.executeAfter,
         createdAt: schema.toolCalls.createdAt,
       })
       .from(schema.toolCalls)
@@ -75,7 +85,7 @@ export const runs = new Hono<AppEnv>()
           : 0;
 
     c.header("x-accel-buffering", "no");
-    return streamSSE(c, async (stream) => {
+    const response = streamSSE(c, async (stream) => {
       let wake: (() => void) | null = null;
       let pending = true; // read once immediately
       const unsubscribe = bus.subscribe(run.id, () => {
@@ -127,6 +137,10 @@ export const runs = new Hono<AppEnv>()
         clearInterval(ping);
       }
     });
+    // A compressing proxy (Next's rewrites gzip by default) buffers small writes, so a
+    // single status change could sit unsent until more output came. no-transform opts out.
+    response.headers.set("cache-control", "no-cache, no-transform");
+    return response;
   })
 
   .post("/:id/tool-calls/:toolCallId/decision", async (c) => {
@@ -141,18 +155,99 @@ export const runs = new Hono<AppEnv>()
     const runId = c.req.param("id");
     if (input.approved && input.input !== undefined)
       await assertEditAllowed(c, runId, c.req.param("toolCallId"), input.input);
-    await decideToolCall(c.var.services.database.db, {
+    const toolCallId = c.req.param("toolCallId");
+    const decided = await decideToolCall(c.var.services.database.db, {
       userId: c.var.user.id,
       runId,
-      toolCallId: c.req.param("toolCallId"),
+      toolCallId,
       approved: input.approved,
       ...(input.reason ? { reason: input.reason } : {}),
       ...(input.input !== undefined ? { editedInput: input.input } : {}),
     });
-    await c.var.services.wake(runId);
+    const { services } = c.var;
+    const { executeAfter } = decided;
+    if (executeAfter) {
+      // Live viewers (another tab, the run panel) get the countdown too.
+      await emitNow(services, runId, {
+        type: "tool.status",
+        toolCallId,
+        status: "approved",
+        executeAfter: executeAfter.toISOString(),
+      });
+    }
+    if (decided.requeued && decided.runnableAt) {
+      // Parked until the window ends: it is queued, not waiting on you.
+      await emitNow(services, runId, { type: "run.status", status: "queued" });
+      wakeAt(services, runId, decided.runnableAt);
+    } else await services.wake(runId);
+    return c.json({ executeAfter: executeAfter?.toISOString() ?? null });
+  })
+
+  .post("/:id/tool-calls/:toolCallId/undo", async (c) => {
+    const runId = c.req.param("id");
+    const toolCallId = c.req.param("toolCallId");
+    const { services } = c.var;
+    await undoToolCall(services.database.db, { userId: c.var.user.id, runId, toolCallId });
+    const output = { type: "denied" as const, reason: UNDO_DENIAL };
+    await emitNow(services, runId, { type: "tool.status", toolCallId, status: "denied" });
+    await emitNow(services, runId, { type: "tool.result", toolCallId, output });
+    await services.wake(runId);
     return c.body(null, 204);
   })
 
+  // Several calls from one run decided at once (the grouped checklist). All or nothing.
+  .post("/:id/tool-calls/decisions", async (c) => {
+    const { decisions } = await body(
+      c,
+      z.object({
+        decisions: z
+          .array(
+            z.object({
+              toolCallId: z.string().min(1),
+              approved: z.boolean(),
+              reason: z.string().max(1000).optional(),
+              input: z.json().optional(),
+            }),
+          )
+          .min(1)
+          .max(100),
+      }),
+    );
+    const runId = c.req.param("id");
+    if (new Set(decisions.map((d) => d.toolCallId)).size !== decisions.length)
+      apiError(400, "duplicate", "Each action can be decided once.");
+    for (const d of decisions)
+      if (d.approved && d.input !== undefined)
+        await assertEditAllowed(c, runId, d.toolCallId, d.input);
+    const { services } = c.var;
+    const decided = await decideToolCalls(services.database.db, {
+      userId: c.var.user.id,
+      runId,
+      decisions: decisions.map((d) => ({
+        toolCallId: d.toolCallId,
+        approved: d.approved,
+        ...(d.reason ? { reason: d.reason } : {}),
+        ...(d.approved && d.input !== undefined ? { editedInput: d.input } : {}),
+      })),
+    });
+    for (const [toolCallId, at] of Object.entries(decided.executeAfter))
+      await emitNow(services, runId, {
+        type: "tool.status",
+        toolCallId,
+        status: "approved",
+        executeAfter: at.toISOString(),
+      });
+    if (decided.requeued && decided.runnableAt) {
+      await emitNow(services, runId, { type: "run.status", status: "queued" });
+      wakeAt(services, runId, decided.runnableAt);
+    } else await services.wake(runId);
+    return c.json({
+      executeAfter: Object.fromEntries(
+        Object.entries(decided.executeAfter).map(([id, at]) => [id, at.toISOString()]),
+      ),
+    });
+  })
+
   .post("/:id/cancel", async (c) => {
     const { mode } = await body(
       c,
@@ -191,6 +286,24 @@ export const approvals = new Hono<AppEnv>().get("/", async (c) => {
   return c.json({ approvals: rows });
 });
 
+/**
+ * Appends an event for a run no worker may be driving (it is parked). Forgets the cached
+ * seq afterwards so a later writer on this instance re-reads the tail.
+ */
+async function emitNow(services: Services, runId: string, event: RunEvent) {
+  await services.events.emit(runId, event);
+  services.events.forget(runId);
+}
+
+/** Wakes workers when a run parked on an undo window becomes claimable, not on the next poll. */
+function wakeAt(services: Services, runId: string, at: Date) {
+  const timer = setTimeout(
+    () => void services.wake(runId).catch(() => {}),
+    Math.max(0, at.getTime() - Date.now()) + 25,
+  );
+  timer.unref?.();
+}
+
 /** Edited arguments are re-checked against block rules: editing must not route around them. */
 async function assertEditAllowed(
   c: { var: AppEnv["Variables"] },

apps/server/src/routes/suggestions.ts

@@ -0,0 +1,99 @@
+import { type PastDecision, SUGGEST_LOOKBACK_DAYS, suggestRules } from "@synapse/core";
+import { PgRunStore, schema } from "@synapse/db";
+import { and, desc, eq, gte, isNotNull, ne, sql } from "drizzle-orm";
+import { Hono } from "hono";
+import { type AppEnv, apiError } from "../http";
+
+type Ctx = { var: AppEnv["Variables"] };
+
+/** Learned rules: offer an allow rule for actions the user keeps approving unchanged. */
+export const suggestions = new Hono<AppEnv>()
+  .get("/", async (c) => {
+    const { suggestions } = await load(c);
+    return c.json({ suggestions });
+  })
+  .post("/:id/accept", async (c) => {
+    const { suggestions, store } = await load(c);
+    const s = suggestions.find((x) => x.id === c.req.param("id"));
+    if (!s) apiError(404, "not_found", "Suggestion not found.");
+    const { db } = c.var.services.database;
+    const userId = c.var.user.id;
+    // First among allow/ask rules so it takes effect; block rules still win regardless.
+    await db.transaction(async (tx) => {
+      await tx
+        .update(schema.policyRules)
+        .set({ position: sql`${schema.policyRules.position} + 1` })
+        .where(eq(schema.policyRules.userId, userId));
+      await tx.insert(schema.policyRules).values({
+        userId,
+        position: 0,
+        tool: s.tool,
+        decision: "allow",
+        when: s.when.length > 0 ? s.when : null,
+        note: s.note,
+      });
+    });
+    return c.json(await store.getPolicy(userId));
+  })
+  .post("/:id/dismiss", async (c) => {
+    const id = c.req.param("id");
+    const { suggestions, dismissed } = await load(c);
+    if (!suggestions.some((x) => x.id === id) && !dismissed.includes(id))
+      apiError(404, "not_found", "Suggestion not found.");
+    // Keep the list bounded; the oldest dismissals fall off first.
+    const next = [...dismissed.filter((d) => d !== id), id].slice(-200);
+    await c.var.services.database.db
+      .insert(schema.userSettings)
+      .values({ userId: c.var.user.id, dismissedSuggestions: next })
+      .onConflictDoUpdate({
+        target: schema.userSettings.userId,
+        set: { dismissedSuggestions: next },
+      });
+    return c.body(null, 204);
+  });
+
+async function load(c: Ctx) {
+  const { db } = c.var.services.database;
+  const userId = c.var.user.id;
+  const since = new Date(Date.now() - SUGGEST_LOOKBACK_DAYS * 24 * 3600_000);
+  const rows = await db
+    .select({
+      tool: schema.toolCalls.name,
+      effect: schema.toolCalls.effect,
+      input: schema.toolCalls.input,
+      inputEdited: schema.toolCalls.inputEdited,
+      status: schema.toolCalls.status,
+      decidedAt: schema.toolCalls.decidedAt,
+      preview: schema.toolCalls.preview,
+    })
+    .from(schema.toolCalls)
+    .where(
+      and(
+        eq(schema.toolCalls.userId, userId),
+        // Human decisions only; auto-allowed calls carry no decider.
+        eq(schema.toolCalls.decidedBy, userId),
+        isNotNull(schema.toolCalls.decidedAt),
+        gte(schema.toolCalls.decidedAt, since),
+        ne(schema.toolCalls.effect, "read"),
+      ),
+    )
+    .orderBy(desc(schema.toolCalls.decidedAt))
+    .limit(2000);
+  const decisions: PastDecision[] = rows.map((r) => ({
+    tool: r.tool,
+    effect: r.effect,
+    input: r.input,
+    inputEdited: r.inputEdited,
+    approved: r.status !== "denied",
+    decidedAt: r.decidedAt!,
+    preview: r.preview,
+  }));
+  const store = new PgRunStore(db);
+  const policy = await store.getPolicy(userId);
+  const [settings] = await db
+    .select({ dismissed: schema.userSettings.dismissedSuggestions })
+    .from(schema.userSettings)
+    .where(eq(schema.userSettings.userId, userId));
+  const dismissed = settings?.dismissed ?? [];
+  return { suggestions: suggestRules(decisions, policy.rules, dismissed), dismissed, store };
+}