diff --git a/docs/workers.md b/docs/workers.md
index 9880ad6..bae2027 100644
--- a/docs/workers.md
+++ b/docs/workers.md
@@ -27,6 +27,7 @@ const worker = new Worker({
   retention: {
     completedMs: 7 * 86_400_000,
     failedMs: 30 * 86_400_000,
+    eventsMs: 86_400_000,
     intervalMs: 600_000,
     batchSize: 100,
   },
@@ -42,7 +43,7 @@ By default a worker claims only its registered types. Separate workers with dist
 
 Worker events are local observations, emitted after corresponding commits where applicable. Listener exceptions and rejected Promises are contained. Synchronous expensive listeners still block the event loop. Reaper dead letters appear in persisted event history, not necessarily a local worker callback. Use durable activities for required follow-up work rather than relying on an event listener.
 
-Retention is disabled by default. It deletes complete terminal trees and their inputs, results, checkpoints, events, keys and dependencies atomically. Live consumer trees pin shared producer results. Completion/failure retention clocks are separate; zero keeps that class forever.
+Retention is disabled by default. It deletes complete terminal trees and their inputs, results, checkpoints, events, keys and dependencies atomically. Live consumer trees pin shared producer results. Completion/failure retention clocks are separate; zero keeps that class forever. `eventsMs` trims the events of finished activities sooner than their tree; zero keeps them with the tree.
 
 ### Executor snapshots
 
diff --git a/spec b/spec
index cf436b7..b24225d 160000
--- a/spec
+++ b/spec
@@ -1 +1 @@
-Subproject commit cf436b7242d33dd2b0c465f9aa5a4f8524484521
+Subproject commit b24225dc4950a2f8c319e5854a5bbae6f1134318
diff --git a/src/postgres/command.ts b/src/postgres/command.ts
index 7170f7d..6e7225d 100644
--- a/src/postgres/command.ts
+++ b/src/postgres/command.ts
@@ -634,7 +634,8 @@ export class PostgresCommands {
       return skipped(row, "still running: cancel it first");
     const live = await c.query(
       `SELECT EXISTS(SELECT 1 FROM runnerq_activities
-      WHERE queue_name=$1 AND root_activity_id=$2 AND status NOT IN ${terminalSQL}) AS live`,
+      WHERE queue_name=$1 AND root_activity_id=$2 AND parent_activity_id IS NOT NULL
+      AND status NOT IN ${terminalSQL}) AS live`,
       [queue, row.id],
     );
     if (live.rows[0].live)
diff --git a/src/postgres/index.ts b/src/postgres/index.ts
index 996c875..c76496c 100644
--- a/src/postgres/index.ts
+++ b/src/postgres/index.ts
@@ -54,7 +54,13 @@ import type {
 } from "../query.js";
 import { Notifications } from "./notifications.js";
 import { PostgresCommands } from "./command.js";
-import { deleteTree, lockTree, terminalSQL } from "./trees.js";
+import {
+  deleteTree,
+  expiredRootSQL,
+  lockExpiredRootSQL,
+  lockTree,
+  trimEventsSQL,
+} from "./trees.js";
 import { attemptsRemain, retryDelaySeconds } from "../retry.js";
 
 export interface PostgresConfig {
@@ -69,6 +75,9 @@ const lost = () =>
   new RunnerQError("claim_lost", "Execution no longer owns this activity");
 const iso = (v: Date | string | null): string | null =>
   v === null ? null : new Date(v).toISOString();
+/** The started_at an event ending an attempt records. Every claim sets it. */
+const attemptStart = (v: Date | null): string =>
+  (v ?? new Date(0)).toISOString();
 function poolConfig(config: Omit<PostgresConfig, "queue">): PoolConfig {
   return {
     connectionString: config.connectionString,
@@ -99,8 +108,11 @@ const addedSince = {
   indexes: [
     "idx_runnerq_commands_created",
     "idx_runnerq_query_created",
-    "idx_runnerq_query_status_created",
-    "idx_runnerq_query_type_created",
+    "idx_runnerq_query_status",
+    "idx_runnerq_processing",
+    "idx_runnerq_root_children",
+    "idx_runnerq_root_terminal",
+    "idx_runnerq_results_by_owner",
   ],
 };
 /**
@@ -170,7 +182,7 @@ async function verifySchema(
   if (retired && !allowOlder)
     throw new RunnerQError(
       "configuration",
-      `Index ${retired.relname} is replaced in this schema version; ${initializeHint}`,
+      `Index ${retired.relname} is retired in this schema version; ${initializeHint}`,
     );
   const keys = await client.query(
     `SELECT c.relname,array_agg(a.attname::text ORDER BY k.ordinality) AS columns FROM pg_constraint p
@@ -208,7 +220,7 @@ async function retryDeadlock(op: () => Promise<unknown>): Promise<void> {
 /**
  * Builds the spec's concurrent indexes one statement at a time (CONCURRENTLY can't run in a
  * transaction): an invalid leftover is dropped and rebuilt, and a replaced index is dropped only
- * once its successor is valid.
+ * once its successor is valid. Then drops the retired indexes.
  */
 async function ensureConcurrentIndexes(client: PoolClient): Promise<void> {
   const schemaName: string = (
@@ -230,6 +242,11 @@ async function ensureConcurrentIndexes(client: PoolClient): Promise<void> {
       if (idx.replaces)
         await client.query(`DROP INDEX IF EXISTS ${ident(idx.replaces)}`);
     });
+  // Retired indexes go once their successors are built.
+  for (const name of catalog.retired.indexes)
+    await retryDeadlock(() =>
+      client.query(`DROP INDEX IF EXISTS ${ident(name)}`),
+    );
 }
 
 export class PostgresStorage
@@ -512,15 +529,14 @@ export class PostgresStorage
       const o = a.options;
       const priority =
         ["low", "normal", "high", "critical"].indexOf(o.priority) + 1;
-      // One statement for the activity, its input, the parent link and the event.
+      // One statement for the activity, its input and the parent link.
       await c.query(
         `WITH activity AS (INSERT INTO runnerq_activities(id,queue_name,activity_type,priority,status,scheduled_at,max_retries,
         timeout_seconds,retry_delay_seconds,max_retry_delay_seconds,metadata,idempotency_key,parent_activity_id,root_activity_id,depth)
         VALUES($1,$2,$3,$4,$5,CASE WHEN $6::bigint>0 THEN NOW()+$6*INTERVAL '1 millisecond' ELSE NULL END,$7,$8,1,$9,$10::jsonb,$11,$12,$13,$14)),
-        input AS (INSERT INTO runnerq_inputs(activity_id,queue_name,payload,serialization) VALUES($1,$2,$15::jsonb,$16)),
         link AS (INSERT INTO runnerq_dependencies(queue_name,waiter_activity_id,result_id,producer_activity_id)
         SELECT $2,$12,$1,$1 WHERE $12::uuid IS NOT NULL ON CONFLICT DO NOTHING)
-        INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail) VALUES($2,$1,$17,NULL,$18::jsonb)`,
+        INSERT INTO runnerq_inputs(activity_id,queue_name,payload,serialization) VALUES($1,$2,$15::jsonb,$16)`,
         [
           a.id,
           this.queue,
@@ -538,8 +554,6 @@ export class PostgresStorage
           a.depth,
           JSON.stringify(a.payload),
           a.serialization,
-          o.delayMs > 0 ? "Scheduled" : "Enqueued",
-          JSON.stringify({ activity_type: a.type, priority }),
         ],
       );
       return a.id;
@@ -567,8 +581,8 @@ export class PostgresStorage
         types.length === 1
           ? "activity_type=$5"
           : "activity_type=ANY($5::text[])";
-      // One statement claims, logs Dequeued and reads the inputs (one round trip for the
-      // batch); the order is the claim order, which UPDATE RETURNING alone doesn't keep.
+      // One statement claims and reads the inputs (one round trip for the batch); the
+      // order is the claim order, which UPDATE RETURNING alone doesn't keep.
       const r = await c.query(
         `WITH picked AS (SELECT id FROM runnerq_activities
         WHERE queue_name=$1 AND status IN ('pending','scheduled','retrying','waiting')
@@ -578,10 +592,7 @@ export class PostgresStorage
         claimed AS (UPDATE runnerq_activities a SET status='processing',current_worker_id=$3||':'||a.id::text,
         started_at=NOW(),waiting_result_id=NULL,
         lease_deadline_ms=(EXTRACT(EPOCH FROM NOW())*1000)::bigint+GREATEST($4::bigint,(timeout_seconds+10)*1000)
-        FROM picked WHERE a.id=picked.id RETURNING a.*),
-        dequeued AS (INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail)
-        SELECT $1,id,'Dequeued',current_worker_id,jsonb_build_object('activity_type',activity_type) FROM claimed
-        ORDER BY priority DESC,retry_count DESC,COALESCE(scheduled_at,created_at) ASC)
+        FROM picked WHERE a.id=picked.id RETURNING a.*)
         SELECT c.id,c.activity_type,c.current_worker_id,c.scheduled_at,c.created_at,c.retry_count,c.timeout_seconds,
         c.parent_activity_id,c.root_activity_id,c.depth,c.metadata,c.lease_deadline_ms,
         i.activity_id AS input_id,i.payload,i.serialization
@@ -660,9 +671,7 @@ export class PostgresStorage
         stored AS (INSERT INTO runnerq_results(activity_id,queue_name,state,data,owner_activity_id,step,serialization)
         SELECT id,$1,'Ok',$4::jsonb,id,NULL,$5 FROM done
         ON CONFLICT(activity_id) DO UPDATE SET data=excluded.data,state=excluded.state,serialization=excluded.serialization,created_at=NOW(),step=excluded.step
-        WHERE runnerq_results.queue_name=excluded.queue_name AND runnerq_results.owner_activity_id=excluded.owner_activity_id),
-        logged AS (INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail)
-        SELECT $1,id,'Completed',$3,'{"result_stored":true}'::jsonb FROM done)
+        WHERE runnerq_results.queue_name=excluded.queue_name AND runnerq_results.owner_activity_id=excluded.owner_activity_id)
         SELECT id FROM done`,
         [this.queue, f.ownerId, f.token, data, value.serialization],
       );
@@ -743,12 +752,15 @@ export class PostgresStorage
         retry_count=retry_count+CASE WHEN $4='retrying' THEN 1 ELSE 0 END,
         scheduled_at=CASE WHEN $4='retrying' THEN NOW()+$6*INTERVAL '1 second' ELSE scheduled_at END,
         started_at=CASE WHEN $4='retrying' THEN NULL ELSE started_at END,
-        completed_at=CASE WHEN $4='retrying' THEN NULL ELSE NOW() END WHERE queue_name=$1 AND id=$2),
+        completed_at=CASE WHEN $4='retrying' THEN NULL ELSE NOW() END WHERE queue_name=$1 AND id=$2
+        RETURNING retry_count,scheduled_at),
         stored AS (INSERT INTO runnerq_results(activity_id,queue_name,state,data,owner_activity_id,step,serialization)
         SELECT $2::uuid,$1,'Err',$7::jsonb,$2::uuid,NULL,'json-v1' WHERE $7::jsonb IS NOT NULL
         ON CONFLICT(activity_id) DO UPDATE SET data=excluded.data,state=excluded.state,serialization=excluded.serialization,created_at=NOW(),step=excluded.step
         WHERE runnerq_results.queue_name=excluded.queue_name AND runnerq_results.owner_activity_id=excluded.owner_activity_id)
-        INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail) VALUES($1,$2,$8,$3,$9::jsonb)`,
+        INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail) SELECT $1,$2,$8,$3,
+        $9::jsonb||CASE $4 WHEN 'retrying' THEN jsonb_build_object('retry_count',retry_count,'scheduled_at',scheduled_at)
+        WHEN 'dead_letter' THEN '{"reason":"attempts_exhausted"}'::jsonb ELSE '{}'::jsonb END FROM failed`,
         [
           this.queue,
           f.ownerId,
@@ -759,8 +771,8 @@ export class PostgresStorage
           result,
           again ? "Retrying" : retry ? "DeadLetter" : "Failed",
           JSON.stringify({
+            started_at: attemptStart(a.started_at),
             error: reason,
-            retryable: retry,
             ...(failure ? { failure } : {}),
           }),
         ],
@@ -778,16 +790,14 @@ export class PostgresStorage
     step: string,
   ): Promise<void> {
     await this.tx(async (c) => {
-      // The fence, the result and (only when the result is new) its ResultStored event:
-      // one statement. The fence's row lock is taken first, as a statement of its own would.
+      // The fence and the result in one statement. The fence's row lock is taken first, as a
+      // statement of its own would.
       const r = await c.query(
         `WITH fenced AS (SELECT id FROM runnerq_activities
         WHERE queue_name=$2 AND id=$5 AND status='processing' AND current_worker_id=$8 FOR UPDATE),
         stored AS (INSERT INTO runnerq_results(activity_id,queue_name,state,data,owner_activity_id,step,serialization)
         SELECT $1::uuid,$2,$3,$4::jsonb,$5::uuid,NULLIF($6,''),$7 WHERE EXISTS(SELECT 1 FROM fenced)
-        ON CONFLICT(activity_id) DO NOTHING RETURNING activity_id),
-        logged AS (INSERT INTO runnerq_events(queue_name,activity_id,event_type,worker_id,detail)
-        SELECT $2,activity_id,'ResultStored',$8,$9::jsonb FROM stored)
+        ON CONFLICT(activity_id) DO NOTHING RETURNING activity_id)
         SELECT EXISTS(SELECT 1 FROM fenced) AS fenced,EXISTS(SELECT 1 FROM stored) AS stored`,
         [
           id,
@@ -798,7 +808,6 @@ export class PostgresStorage
           step,
           result.serialization,
           f.token,
-          JSON.stringify({ state: result.state }),
         ],
       );
       if (!r.rows[0].fenced) throw lost();
@@ -984,15 +993,18 @@ export class PostgresStorage
   }
   async reap(limit: number): Promise<number> {
     const ids = await this.tx(async (c) => {
+      // The expired attempt's worker and start come from the locked rows, before the update
+      // clears them, for the events.
       const rows = await c.query(
-        `UPDATE runnerq_activities SET retry_count=retry_count+CASE WHEN max_retries>0 AND retry_count+1>=max_retries THEN 0 ELSE 1 END,
-        status=CASE WHEN max_retries>0 AND retry_count+1>=max_retries THEN 'dead_letter' ELSE 'pending' END,
-        completed_at=CASE WHEN max_retries>0 AND retry_count+1>=max_retries THEN NOW() ELSE NULL END,
+        `UPDATE runnerq_activities a SET retry_count=a.retry_count+CASE WHEN a.max_retries>0 AND a.retry_count+1>=a.max_retries THEN 0 ELSE 1 END,
+        status=CASE WHEN a.max_retries>0 AND a.retry_count+1>=a.max_retries THEN 'dead_letter' ELSE 'pending' END,
+        completed_at=CASE WHEN a.max_retries>0 AND a.retry_count+1>=a.max_retries THEN NOW() ELSE NULL END,
         last_error='lease expired before completion; worker presumed crashed or wedged',last_error_at=NOW(),
-        last_worker_id=current_worker_id,current_worker_id=NULL,lease_deadline_ms=NULL,started_at=NULL,waiting_result_id=NULL
-        WHERE id IN (SELECT id FROM runnerq_activities WHERE queue_name=$1 AND status='processing'
-          AND lease_deadline_ms<(EXTRACT(EPOCH FROM NOW())*1000)::bigint LIMIT $2 FOR UPDATE SKIP LOCKED)
-        RETURNING id,status,last_error`,
+        last_worker_id=a.current_worker_id,current_worker_id=NULL,lease_deadline_ms=NULL,started_at=NULL,waiting_result_id=NULL
+        FROM (SELECT id,current_worker_id,started_at FROM runnerq_activities WHERE queue_name=$1 AND status='processing'
+          AND lease_deadline_ms<(EXTRACT(EPOCH FROM NOW())*1000)::bigint LIMIT $2 FOR UPDATE SKIP LOCKED) expired
+        WHERE a.id=expired.id
+        RETURNING a.id,a.status,a.last_error,a.retry_count,expired.current_worker_id AS worker,expired.started_at AS attempt_started`,
         [this.queue, integer(limit, "reaper limit", 1)],
       );
       for (const a of rows.rows) {
@@ -1016,8 +1028,19 @@ export class PostgresStorage
           c,
           a.id,
           a.status === "dead_letter" ? "DeadLetter" : "Requeued",
-          null,
-          { reason: "lease_expired", error: a.last_error },
+          a.worker,
+          a.status === "dead_letter"
+            ? {
+                started_at: attemptStart(a.attempt_started),
+                reason: "lease_expired",
+                error: a.last_error,
+              }
+            : {
+                started_at: attemptStart(a.attempt_started),
+                reason: "lease_expired",
+                error: a.last_error,
+                retry_count: a.retry_count,
+              },
         );
       }
       return rows.rows;
@@ -1028,9 +1051,10 @@ export class PostgresStorage
   }
   async cleanup(policy: Retention): Promise<number> {
     const completed = integer(policy.completedMs ?? 0, "completedMs"),
-      failed = integer(policy.failedMs ?? 0, "failedMs");
+      failed = integer(policy.failedMs ?? 0, "failedMs"),
+      events = integer(policy.eventsMs ?? 0, "eventsMs");
     const batch = integer(policy.batchSize ?? 100, "batchSize", 1);
-    if (!completed && !failed) return 0;
+    if (!completed && !failed && !events) return 0;
     return this.tx(async (c) => {
       if (
         !(
@@ -1050,19 +1074,29 @@ export class PostgresStorage
         examined++
       ) {
         await c.query("SAVEPOINT candidate");
-        const r = await c.query(
-          `SELECT r.id FROM runnerq_activities r WHERE r.queue_name=$1 AND r.parent_activity_id IS NULL
-          AND r.id<>ALL($4::uuid[]) AND ((r.status='completed' AND $2::bigint>0 AND r.completed_at<NOW()-$2*INTERVAL '1 millisecond')
-          OR (r.status IN ('failed','dead_letter','cancelled') AND $3::bigint>0 AND r.completed_at<NOW()-$3*INTERVAL '1 millisecond'))
-          AND NOT EXISTS(SELECT 1 FROM runnerq_activities a WHERE a.queue_name=$1 AND a.root_activity_id=r.id AND a.status NOT IN ${terminalSQL})
-          ORDER BY r.completed_at LIMIT 1 FOR UPDATE SKIP LOCKED`,
-          [this.queue, completed, failed, skipped],
-        );
+        const r = await c.query(expiredRootSQL, [
+          this.queue,
+          completed,
+          failed,
+          skipped,
+        ]);
         const root = r.rows[0]?.id;
         if (!root) {
           await c.query("RELEASE SAVEPOINT candidate");
           break;
         }
+        // The candidate was read unlocked: lock it, and recheck it, now.
+        const locked = await c.query(lockExpiredRootSQL, [
+          this.queue,
+          completed,
+          failed,
+          root,
+        ]);
+        if (!locked.rows[0].locked) {
+          skipped.push(root);
+          await c.query("RELEASE SAVEPOINT candidate");
+          continue;
+        }
         if (await lockTree(c, this.queue, root)) {
           skipped.push(root);
           await c.query("ROLLBACK TO SAVEPOINT candidate");
@@ -1073,6 +1107,8 @@ export class PostgresStorage
         await c.query("RELEASE SAVEPOINT candidate");
         removed++;
       }
+      if (events)
+        await c.query(trimEventsSQL, [this.queue, events, batch * 10]);
       return removed;
     });
   }
@@ -1110,7 +1146,6 @@ export class PostgresStorage
     for (const [column, value] of [
       ["status", options.status],
       ["parent_activity_id", options.parentId],
-      ["root_activity_id", options.rootId],
       ["metadata->>'source'", options.source],
     ]) {
       if (value !== undefined) {
@@ -1118,6 +1153,13 @@ export class PostgresStorage
         predicates.push(`${column}=$${values.length}`);
       }
     }
+    if (options.rootId !== undefined) {
+      values.push(options.rootId);
+      const p = `$${values.length}`;
+      predicates.push(
+        `(id=${p} OR (root_activity_id=${p} AND parent_activity_id IS NOT NULL))`,
+      );
+    }
     values.push(
       integer(options.limit ?? 50, "limit", 1, 1000),
       integer(options.offset ?? 0, "offset"),
diff --git a/src/postgres/query.ts b/src/postgres/query.ts
index 73dd8f4..3ba5de9 100644
--- a/src/postgres/query.ts
+++ b/src/postgres/query.ts
@@ -60,7 +60,7 @@ const activityFields: Record<string, Field> = {
   queue: { expr: "a.queue_name", kind: "string" },
   status: { expr: "a.status", kind: "status" },
   priority: { expr: "a.priority", kind: "int" },
-  root_id: { expr: "COALESCE(a.root_activity_id, a.id)", kind: "uuid" },
+  root_id: { expr: "a.root_activity_id", kind: "uuid" },
   parent_id: { expr: "a.parent_activity_id", kind: "uuid", nullable: true },
   depth: { expr: "a.depth", kind: "int" },
   idempotency_key: {
@@ -134,18 +134,12 @@ export function canonicalStatus(internal: string): RecordStatus {
 
 /** Internal event names (Go's and TypeScript's are the same) to canonical event types. */
 const canonicalEvents: Record<string, string> = {
-  Enqueued: RecordEvent.created,
-  Scheduled: RecordEvent.scheduled,
-  Dequeued: RecordEvent.attemptStarted,
-  Completed: RecordEvent.attemptSucceeded,
   Failed: RecordEvent.attemptFailed,
   Retrying: RecordEvent.attemptFailed,
   DeadLetter: RecordEvent.deadLetter,
   Requeued: RecordEvent.leaseExpired,
   Yielded: RecordEvent.waitParked,
   Signaled: RecordEvent.signalReceived,
-  LeaseExtended: RecordEvent.leaseExtended,
-  ResultStored: RecordEvent.resultStored,
   SpawnLinked: RecordEvent.childLinked,
   Cancelled: RecordEvent.cancelled,
   Retried: RecordEvent.retried,
@@ -270,7 +264,17 @@ export class SqlBuilder {
         const negate = f.op === "ne" || f.op === "nin";
         // Nothing can match (e.g. an id that is not a UUID).
         if (!vals) return String(negate);
-        const cond = `${fd.expr} = ANY(${this.arg(vals.values, vals.type + "[]")})`;
+        let cond: string;
+        if (fd.kind === "status")
+          // Literals, not a parameter: Postgres proves a partial index's predicate
+          // (idx_runnerq_query_status) only from constants. The values are canonicalStatuses'.
+          cond = `${fd.expr} IN (${(vals.values as string[]).map((v) => `'${v.replaceAll("'", "''")}'`).join(", ")})`;
+        else if (name === "root_id") {
+          // Roots aren't in idx_runnerq_root_children; they're found by id.
+          const p = this.arg(vals.values, "uuid[]");
+          cond = `(a.id = ANY(${p}) OR (a.root_activity_id = ANY(${p}) AND a.parent_activity_id IS NOT NULL))`;
+        } else
+          cond = `${fd.expr} = ANY(${this.arg(vals.values, vals.type + "[]")})`;
         return negate ? `(NOT COALESCE(${cond}, false))` : cond;
       }
       case "lt":
@@ -348,7 +352,7 @@ export class SqlBuilder {
     const vals = convertValues(name, "uuid", raw);
     if (!vals) return "false";
     const p = this.arg(vals.values, "uuid[]");
-    return `e.activity_id IN (SELECT x.id FROM runnerq_activities x WHERE x.id = ANY(${p}) OR x.root_activity_id = ANY(${p}))`;
+    return `e.activity_id IN (SELECT x.id FROM runnerq_activities x WHERE x.id = ANY(${p}) OR (x.root_activity_id = ANY(${p}) AND x.parent_activity_id IS NOT NULL))`;
   }
 }
 
@@ -934,7 +938,7 @@ export class PostgresQueries {
     const { cols, joins } = activitySelect(inc);
     const rows = await this.run(
       `SELECT ${cols} FROM runnerq_activities a${joins}
-      WHERE a.id = $1::uuid OR a.root_activity_id = $1::uuid
+      WHERE a.id = $1::uuid OR (a.root_activity_id = $1::uuid AND a.parent_activity_id IS NOT NULL)
       ORDER BY a.depth, a.created_at, a.id LIMIT ${max + 1}`,
       [root],
     );
diff --git a/src/postgres/trees.ts b/src/postgres/trees.ts
index a9e1c67..09a862d 100644
--- a/src/postgres/trees.ts
+++ b/src/postgres/trees.ts
@@ -5,6 +5,13 @@ import type { PoolClient } from "pg";
 /** The statuses an activity never leaves (bar a retry command). */
 export const terminalSQL = "('completed','failed','dead_letter','cancelled')";
 
+/**
+ * The workflow rooted at $2 in queue $1: the root by id, its descendants through
+ * idx_runnerq_root_children.
+ */
+export const treeSQL = `SELECT a.id,a.idempotency_key FROM runnerq_activities a WHERE a.queue_name=$1
+  AND (a.id=$2 OR (a.root_activity_id=$2 AND a.parent_activity_id IS NOT NULL))`;
+
 /**
  * Locks the tree's idempotency keys, then reports whether a live workflow in another tree
  * depends on a result this tree produced (the tree must then be kept). Key reuse registers
@@ -15,19 +22,20 @@ export async function lockTree(
   queue: string,
   root: string,
 ): Promise<boolean> {
+  // A key's activity carries the key, so each key is a primary-key probe.
   await c.query(
-    `SELECT 1 FROM runnerq_idempotency WHERE queue_name=$1 AND activity_id IN
-    (SELECT id FROM runnerq_activities WHERE queue_name=$1 AND (id=$2 OR root_activity_id=$2)) FOR UPDATE`,
+    `SELECT 1 FROM runnerq_idempotency k JOIN (${treeSQL}) t
+    ON k.queue_name=$1 AND k.idempotency_key=t.idempotency_key AND k.activity_id=t.id FOR UPDATE OF k`,
     [queue, root],
   );
   const r = await c.query(
-    `SELECT EXISTS(SELECT 1 FROM runnerq_dependencies d
-    JOIN runnerq_activities producer ON producer.id=d.producer_activity_id AND producer.queue_name=$1
+    `SELECT EXISTS(SELECT 1 FROM (${treeSQL}) t
+    JOIN runnerq_dependencies d ON d.queue_name=$1 AND d.result_id=t.id AND d.producer_activity_id=t.id
     JOIN runnerq_activities waiter ON waiter.id=d.waiter_activity_id AND waiter.queue_name=$1
-    WHERE d.queue_name=$1 AND COALESCE(producer.root_activity_id,producer.id)=$2
-    AND COALESCE(waiter.root_activity_id,waiter.id)<>$2
+    WHERE waiter.root_activity_id<>$2
     AND EXISTS(SELECT 1 FROM runnerq_activities live WHERE live.queue_name=$1
-      AND COALESCE(live.root_activity_id,live.id)=COALESCE(waiter.root_activity_id,waiter.id)
+      AND (live.id=waiter.root_activity_id
+        OR (live.root_activity_id=waiter.root_activity_id AND live.parent_activity_id IS NOT NULL))
       AND live.status NOT IN ${terminalSQL})) AS pinned`,
     [queue, root],
   );
@@ -41,16 +49,53 @@ export async function deleteTree(
   root: string,
 ): Promise<void> {
   await c.query(
-    `WITH tree AS (SELECT id FROM runnerq_activities WHERE queue_name=$1 AND (id=$2 OR root_activity_id=$2)),
+    `WITH tree AS (${treeSQL}),
     del_dependencies AS (DELETE FROM runnerq_dependencies WHERE queue_name=$1
-      AND (waiter_activity_id IN (SELECT id FROM tree) OR producer_activity_id IN (SELECT id FROM tree))),
+      AND (waiter_activity_id IN (SELECT id FROM tree)
+        OR (result_id IN (SELECT id FROM tree) AND producer_activity_id IS NOT NULL))),
     del_results AS (DELETE FROM runnerq_results WHERE queue_name=$1
       AND (activity_id IN (SELECT id FROM tree) OR owner_activity_id IN (SELECT id FROM tree)) RETURNING activity_id),
     del_events AS (DELETE FROM runnerq_events WHERE queue_name=$1
       AND (activity_id IN (SELECT id FROM tree) OR activity_id IN (SELECT activity_id FROM del_results))),
-    del_idem AS (DELETE FROM runnerq_idempotency WHERE queue_name=$1 AND activity_id IN (SELECT id FROM tree)),
+    del_idem AS (DELETE FROM runnerq_idempotency k USING tree
+      WHERE k.queue_name=$1 AND k.idempotency_key=tree.idempotency_key AND k.activity_id=tree.id),
     del_inputs AS (DELETE FROM runnerq_inputs WHERE activity_id IN (SELECT id FROM tree))
     DELETE FROM runnerq_activities WHERE queue_name=$1 AND id IN (SELECT id FROM tree)`,
     [queue, root],
   );
 }
+
+/** $2 and $3 are the completed and failed ages in milliseconds (0 keeps them). */
+const expiredCondSQL = `r.status IN ${terminalSQL}
+  AND r.completed_at<NOW()-(CASE WHEN r.status='completed' THEN $2::bigint ELSE $3::bigint END)*INTERVAL '1 millisecond'
+  AND CASE WHEN r.status='completed' THEN $2::bigint ELSE $3::bigint END>0`;
+
+/**
+ * The oldest expired root of queue $1 with no live descendant ($4 lists roots already passed
+ * over): one short walk of idx_runnerq_root_terminal per terminal status.
+ */
+export const expiredRootSQL = `SELECT r.id FROM unnest(ARRAY['completed','failed','dead_letter','cancelled']) s(status)
+  CROSS JOIN LATERAL (SELECT r.id,r.completed_at FROM runnerq_activities r
+    WHERE r.queue_name=$1 AND r.parent_activity_id IS NULL AND r.status IN ${terminalSQL} AND r.status=s.status
+    AND r.completed_at<NOW()-(CASE WHEN s.status='completed' THEN $2::bigint ELSE $3::bigint END)*INTERVAL '1 millisecond'
+    AND r.id<>ALL($4::uuid[])
+    AND NOT EXISTS(SELECT 1 FROM runnerq_activities c WHERE c.root_activity_id=r.id AND c.parent_activity_id IS NOT NULL
+      AND c.status NOT IN ${terminalSQL})
+    ORDER BY r.completed_at LIMIT 1) r
+  WHERE CASE WHEN s.status='completed' THEN $2::bigint ELSE $3::bigint END>0
+  ORDER BY r.completed_at LIMIT 1`;
+
+/** Locks and rechecks a candidate root ($4) that expiredRootSQL read unlocked. */
+export const lockExpiredRootSQL = `SELECT EXISTS(SELECT 1 FROM runnerq_activities r
+  WHERE r.id=$4 AND r.queue_name=$1 AND r.parent_activity_id IS NULL AND ${expiredCondSQL}
+  FOR UPDATE SKIP LOCKED) AS locked`;
+
+/**
+ * Deletes up to $3 events older than $2 milliseconds whose activity has finished (or is
+ * gone), oldest first along idx_runnerq_events_queue_seq. Events of unfinished activities
+ * stay: a waiting activity's latest Yielded is its waiting reason.
+ */
+export const trimEventsSQL = `DELETE FROM runnerq_events WHERE id IN (SELECT e.id FROM runnerq_events e
+  WHERE e.queue_name=$1 AND e.created_at<NOW()-$2::bigint*INTERVAL '1 millisecond'
+  AND NOT EXISTS(SELECT 1 FROM runnerq_activities a WHERE a.id=e.activity_id AND a.status NOT IN ${terminalSQL})
+  ORDER BY e.queue_name,e.id LIMIT $3)`;
diff --git a/src/query.ts b/src/query.ts
index 60a0c58..cfcdb80 100644
--- a/src/query.ts
+++ b/src/query.ts
@@ -62,6 +62,9 @@ export type RecordStatus =
 /**
  * Canonical event types. Backends map their internal event names onto these; events with
  * no canonical equivalent keep a namespaced name of their own ("other.<name>").
+ *
+ * `created`, `scheduled`, `attemptStarted` and `attemptSucceeded` are never stored: an
+ * activity's createdAt, scheduledAt, startedAt and completedAt say them.
  */
 export const RecordEvent = {
   created: "activity.created",
@@ -72,10 +75,8 @@ export const RecordEvent = {
   attemptFailed: "attempt.failed",
   attemptTimedOut: "attempt.timed_out",
   leaseExpired: "attempt.lease_expired",
-  leaseExtended: "attempt.lease_extended",
   waitParked: "wait.parked",
   signalReceived: "signal.received",
-  resultStored: "result.stored",
   childLinked: "child.linked",
   deadLetter: "dead_letter.entered",
   redriven: "dead_letter.redriven",
diff --git a/src/spec-schema.ts b/src/spec-schema.ts
index abc7ede..c1d5782 100644
--- a/src/spec-schema.ts
+++ b/src/spec-schema.ts
@@ -32,7 +32,7 @@ export interface Catalog {
 export const postgresMigrations: readonly PostgresMigration[] = [
   {
     "name": "0001_baseline",
-    "sql": "-- Activities table (permanent - no TTL)\nCREATE TABLE IF NOT EXISTS runnerq_activities (\n    id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    activity_type TEXT NOT NULL,\n    priority INTEGER NOT NULL DEFAULT 1,\n    status TEXT NOT NULL DEFAULT 'pending',\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    scheduled_at TIMESTAMPTZ,\n    started_at TIMESTAMPTZ,\n    completed_at TIMESTAMPTZ,\n    lease_deadline_ms BIGINT,\n    current_worker_id TEXT,\n    last_worker_id TEXT,\n    retry_count INTEGER NOT NULL DEFAULT 0,\n    max_retries INTEGER NOT NULL DEFAULT 0,\n    timeout_seconds BIGINT NOT NULL DEFAULT 300,\n    retry_delay_seconds BIGINT NOT NULL DEFAULT 60,\n    max_retry_delay_seconds BIGINT NOT NULL DEFAULT 0,\n    last_error TEXT,\n    last_error_at TIMESTAMPTZ,\n    metadata JSONB,\n    idempotency_key TEXT\n);\n\n-- Inputs, apart from the activity rows that claims, leases and status\n-- changes rewrite: an activity's payload is written once and read when it is\n-- claimed or inspected. serialization names the payload's encoding\n-- ('json-v1': plain JSON, the only one Go writes; the TypeScript SDK also\n-- writes 'superjson-v1').\nCREATE TABLE IF NOT EXISTS runnerq_inputs (\n    activity_id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    payload JSONB NOT NULL,\n    serialization TEXT NOT NULL DEFAULT 'json-v1'\n);\n\n-- Migration: payloads used to live in runnerq_activities.payload. Move them\n-- into runnerq_inputs and drop the column, in the schema transaction, so a\n-- database is either wholly before or wholly after the move. Processes on\n-- an older version of this SDK can't use the database afterwards: stop them\n-- before starting this one.\nDO $$\nBEGIN\n    IF EXISTS (SELECT 1 FROM information_schema.columns\n               WHERE table_schema = current_schema()\n                 AND table_name = 'runnerq_activities' AND column_name = 'payload') THEN\n        INSERT INTO runnerq_inputs (activity_id, queue_name, payload)\n            SELECT id, queue_name, payload FROM runnerq_activities\n            ON CONFLICT (activity_id) DO NOTHING;\n        ALTER TABLE runnerq_activities DROP COLUMN payload;\n        ALTER TABLE runnerq_activities ALTER COLUMN max_retries SET DEFAULT 0;\n    END IF;\nEND $$;\n\n-- Indexes for efficient queries\n--\n-- NOTE: the hot dequeue indexes (idx_runnerq_dequeue_effective_v2,\n-- idx_runnerq_dequeue_order_v2) and the query indexes are NOT created here:\n-- see concurrent_indexes.json. A plain CREATE INDEX takes a SHARE lock that\n-- blocks all writes on runnerq_activities for the duration of the build; at\n-- boot, on a hot queue, that stalls every enqueue/dequeue/ack in the cluster.\nCREATE INDEX IF NOT EXISTS idx_runnerq_activities_processing\n    ON runnerq_activities(queue_name, lease_deadline_ms)\n    WHERE status = 'processing';\nCREATE INDEX IF NOT EXISTS idx_runnerq_completed_non_cron\n    ON runnerq_activities(queue_name, completed_at DESC, created_at DESC)\n    WHERE status IN ('completed', 'failed')\n      AND (metadata-\u003e\u003e'source') IS DISTINCT FROM 'cron';\nCREATE INDEX IF NOT EXISTS idx_runnerq_completed_cron\n    ON runnerq_activities(queue_name, completed_at DESC, created_at DESC)\n    WHERE status IN ('completed', 'failed')\n      AND metadata-\u003e\u003e'source' = 'cron';\nCREATE INDEX IF NOT EXISTS idx_runnerq_dead_letter\n    ON runnerq_activities(queue_name, completed_at DESC)\n    WHERE status = 'dead_letter';\n\n-- Migration: add column for existing deployments (idempotent, metadata-only on PG 11+).\nALTER TABLE runnerq_activities\n    ADD COLUMN IF NOT EXISTS max_retry_delay_seconds BIGINT NOT NULL DEFAULT 0;\n\n-- Lineage columns for parent/child activity tracking.\nALTER TABLE runnerq_activities\n    ADD COLUMN IF NOT EXISTS parent_activity_id UUID,\n    ADD COLUMN IF NOT EXISTS root_activity_id UUID,\n    ADD COLUMN IF NOT EXISTS depth SMALLINT NOT NULL DEFAULT 0,\n    ADD COLUMN IF NOT EXISTS waiting_result_id UUID;\n\nCREATE INDEX IF NOT EXISTS idx_runnerq_parent_id\n    ON runnerq_activities(parent_activity_id)\n    WHERE parent_activity_id IS NOT NULL;\nCREATE INDEX IF NOT EXISTS idx_runnerq_root_id\n    ON runnerq_activities(root_activity_id)\n    WHERE root_activity_id IS NOT NULL;\n-- Partial index for the workflows-list view (parent IS NULL = roots only),\n-- ordered for the typical \"newest first\" query.\nCREATE INDEX IF NOT EXISTS idx_runnerq_root_only\n    ON runnerq_activities(queue_name, created_at DESC)\n    WHERE parent_activity_id IS NULL;\nCREATE INDEX IF NOT EXISTS idx_runnerq_root_status\n    ON runnerq_activities(queue_name, status)\n    WHERE parent_activity_id IS NULL;\n\n-- Idempotency keys table\nCREATE TABLE IF NOT EXISTS runnerq_idempotency (\n    queue_name TEXT NOT NULL,\n    idempotency_key TEXT NOT NULL,\n    activity_id UUID NOT NULL,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    PRIMARY KEY (queue_name, idempotency_key)\n);\n\n-- Events table (permanent - full history)\nCREATE TABLE IF NOT EXISTS runnerq_events (\n    id BIGSERIAL PRIMARY KEY,\n    activity_id UUID NOT NULL,\n    queue_name TEXT NOT NULL,\n    event_type TEXT NOT NULL,\n    worker_id TEXT,\n    detail JSONB,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\n\nCREATE INDEX IF NOT EXISTS idx_runnerq_events_activity\n    ON runnerq_events(activity_id, created_at DESC);\n-- Cursor tailing for the live event stream (EventStream reads id \u003e cursor).\nCREATE INDEX IF NOT EXISTS idx_runnerq_events_queue_seq\n    ON runnerq_events(queue_name, id);\n\n-- Results table. Rows live until the retention sweeper deletes their\n-- workflow tree (or forever when retention is not configured).\nCREATE TABLE IF NOT EXISTS runnerq_results (\n    activity_id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    state TEXT NOT NULL,\n    data JSONB,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\n\n-- owner_activity_id ties every result row to the activity whose lifetime\n-- governs it. For normal activity results it equals activity_id; for Run/\n-- Sleep checkpoints (whose activity_id is a synthetic UUIDv5 matching no\n-- activity row) it is the handler's activity — without it, checkpoint rows\n-- could never be garbage-collected. NULL on legacy rows (never swept).\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS owner_activity_id UUID;\nCREATE INDEX IF NOT EXISTS idx_runnerq_results_owner\n    ON runnerq_results(queue_name, owner_activity_id)\n    WHERE owner_activity_id IS NOT NULL;\n\n-- step is the human identity of a checkpoint row, \"kind:name\" (e.g.\n-- \"run:create-transfer\", \"sleep:retry-backoff\") — the same string the checkpoint\n-- ID is derived from, persisted so the console can show a workflow's durable\n-- step history (the checkpoint ID itself is a one-way hash). NULL for an\n-- activity's own final result and for legacy rows. No dedicated index: read only\n-- by owner (already indexed above) on the console path, never on a hot path.\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS step TEXT;\n\n-- serialization names the result data's encoding, as for inputs.\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS serialization TEXT NOT NULL DEFAULT 'json-v1';\n\n-- Retention sweep: find roots that have been terminal longer than the TTL.\nCREATE INDEX IF NOT EXISTS idx_runnerq_root_terminal_age\n    ON runnerq_activities(queue_name, status, completed_at)\n    WHERE parent_activity_id IS NULL\n      AND status IN ('completed', 'failed', 'dead_letter');\n\n-- Worker pools: one row per live engine instance. Heartbeats let us tell\n-- which pools are still alive for cluster-wide capacity reporting.\nCREATE TABLE IF NOT EXISTS runnerq_worker_pools (\n    pool_id        UUID PRIMARY KEY,\n    queue_name     TEXT NOT NULL,\n    max_workers    INTEGER NOT NULL,\n    activity_types TEXT[],\n    started_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    last_seen_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_worker_pools_queue_alive\n    ON runnerq_worker_pools(queue_name, last_seen_at);\n\n-- Command ledger: applied RunnerQ Cloud commands and their results, so a\n-- repeated command_id replays its result instead of applying twice. Rows are\n-- pruned after a week (commands must be replayable for at least 24 hours).\nCREATE TABLE IF NOT EXISTS runnerq_commands (\n    queue_name  TEXT NOT NULL,\n    command_id  TEXT NOT NULL,\n    fingerprint TEXT NOT NULL,\n    kind        TEXT NOT NULL,\n    result      JSONB NOT NULL,\n    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    PRIMARY KEY (queue_name, command_id)\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_commands_created\n    ON runnerq_commands(created_at);\n\n-- Durable references survive handler replay and are removed with consumer trees.\nCREATE TABLE IF NOT EXISTS runnerq_dependencies (\n queue_name TEXT NOT NULL,\n waiter_activity_id UUID NOT NULL,\n result_id UUID NOT NULL,\n producer_activity_id UUID,\n created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n PRIMARY KEY (queue_name, waiter_activity_id, result_id)\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_dependencies_result\n ON runnerq_dependencies(queue_name, result_id);\nCREATE INDEX IF NOT EXISTS idx_runnerq_dependencies_producer\n ON runnerq_dependencies(queue_name, producer_activity_id);\n"
+    "sql": "-- Activities table (permanent - no TTL)\nCREATE TABLE IF NOT EXISTS runnerq_activities (\n    id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    activity_type TEXT NOT NULL,\n    priority INTEGER NOT NULL DEFAULT 1,\n    status TEXT NOT NULL DEFAULT 'pending',\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    scheduled_at TIMESTAMPTZ,\n    started_at TIMESTAMPTZ,\n    completed_at TIMESTAMPTZ,\n    lease_deadline_ms BIGINT,\n    current_worker_id TEXT,\n    last_worker_id TEXT,\n    retry_count INTEGER NOT NULL DEFAULT 0,\n    max_retries INTEGER NOT NULL DEFAULT 0,\n    timeout_seconds BIGINT NOT NULL DEFAULT 300,\n    retry_delay_seconds BIGINT NOT NULL DEFAULT 60,\n    max_retry_delay_seconds BIGINT NOT NULL DEFAULT 0,\n    last_error TEXT,\n    last_error_at TIMESTAMPTZ,\n    metadata JSONB,\n    idempotency_key TEXT\n);\n\n-- Inputs, apart from the activity rows that claims, leases and status\n-- changes rewrite: an activity's payload is written once and read when it is\n-- claimed or inspected. serialization names the payload's encoding\n-- ('json-v1': plain JSON, the only one Go writes; the TypeScript SDK also\n-- writes 'superjson-v1').\nCREATE TABLE IF NOT EXISTS runnerq_inputs (\n    activity_id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    payload JSONB NOT NULL,\n    serialization TEXT NOT NULL DEFAULT 'json-v1'\n);\n\n-- Migration: payloads used to live in runnerq_activities.payload. Move them\n-- into runnerq_inputs and drop the column, in the schema transaction, so a\n-- database is either wholly before or wholly after the move. Processes on\n-- an older version of this SDK can't use the database afterwards: stop them\n-- before starting this one.\nDO $$\nBEGIN\n    IF EXISTS (SELECT 1 FROM information_schema.columns\n               WHERE table_schema = current_schema()\n                 AND table_name = 'runnerq_activities' AND column_name = 'payload') THEN\n        INSERT INTO runnerq_inputs (activity_id, queue_name, payload)\n            SELECT id, queue_name, payload FROM runnerq_activities\n            ON CONFLICT (activity_id) DO NOTHING;\n        ALTER TABLE runnerq_activities DROP COLUMN payload;\n        ALTER TABLE runnerq_activities ALTER COLUMN max_retries SET DEFAULT 0;\n    END IF;\nEND $$;\n\n-- Indexes for efficient queries\n--\n-- NOTE: indexes on runnerq_activities and runnerq_results are NOT created\n-- here: see concurrent_indexes.json. A plain CREATE INDEX takes a SHARE lock\n-- that blocks all writes on the table for the duration of the build; at boot,\n-- on a hot queue, that stalls every enqueue/dequeue/ack in the cluster.\n--\n-- Every index on runnerq_activities is written again by each status change,\n-- so an index must earn its place. Room left on each page lets lease renewals,\n-- which change no indexed column, update in place (HOT).\nALTER TABLE runnerq_activities SET (fillfactor = 85);\n\n-- Migration: add column for existing deployments (idempotent, metadata-only on PG 11+).\nALTER TABLE runnerq_activities\n    ADD COLUMN IF NOT EXISTS max_retry_delay_seconds BIGINT NOT NULL DEFAULT 0;\n\n-- Lineage columns for parent/child activity tracking.\nALTER TABLE runnerq_activities\n    ADD COLUMN IF NOT EXISTS parent_activity_id UUID,\n    ADD COLUMN IF NOT EXISTS root_activity_id UUID,\n    ADD COLUMN IF NOT EXISTS depth SMALLINT NOT NULL DEFAULT 0,\n    ADD COLUMN IF NOT EXISTS waiting_result_id UUID;\n\nCREATE INDEX IF NOT EXISTS idx_runnerq_parent_id\n    ON runnerq_activities(parent_activity_id)\n    WHERE parent_activity_id IS NOT NULL;\n\n-- Idempotency keys table\nCREATE TABLE IF NOT EXISTS runnerq_idempotency (\n    queue_name TEXT NOT NULL,\n    idempotency_key TEXT NOT NULL,\n    activity_id UUID NOT NULL,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    PRIMARY KEY (queue_name, idempotency_key)\n);\n\n-- Events: what the activity row can't say (events.schema.json). Retention\n-- removes them with their tree, or sooner with an events window.\nCREATE TABLE IF NOT EXISTS runnerq_events (\n    id BIGSERIAL PRIMARY KEY,\n    activity_id UUID NOT NULL,\n    queue_name TEXT NOT NULL,\n    event_type TEXT NOT NULL,\n    worker_id TEXT,\n    detail JSONB,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\n\nCREATE INDEX IF NOT EXISTS idx_runnerq_events_activity\n    ON runnerq_events(activity_id, created_at DESC);\n-- Per-queue log order: the live stream's tail (id \u003e cursor) and the events\n-- retention walk.\nCREATE INDEX IF NOT EXISTS idx_runnerq_events_queue_seq\n    ON runnerq_events(queue_name, id);\n\n-- Results table. Rows live until the retention sweeper deletes their\n-- workflow tree (or forever when retention is not configured).\nCREATE TABLE IF NOT EXISTS runnerq_results (\n    activity_id UUID PRIMARY KEY,\n    queue_name TEXT NOT NULL,\n    state TEXT NOT NULL,\n    data JSONB,\n    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\n\n-- owner_activity_id ties every result row to the activity whose lifetime\n-- governs it. For normal activity results it equals activity_id; for Run/\n-- Sleep checkpoints (whose activity_id is a synthetic UUIDv5 matching no\n-- activity row) it is the handler's activity — without it, checkpoint rows\n-- could never be garbage-collected. NULL on legacy rows (never swept).\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS owner_activity_id UUID;\n\n-- step is the human identity of a checkpoint row, \"kind:name\" (e.g.\n-- \"run:create-transfer\", \"sleep:retry-backoff\") — the same string the checkpoint\n-- ID is derived from, persisted so the console can show a workflow's durable\n-- step history (the checkpoint ID itself is a one-way hash). NULL for an\n-- activity's own final result and for legacy rows. No dedicated index: read only\n-- by owner (idx_runnerq_results_by_owner) on the console path, never on a hot path.\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS step TEXT;\n\n-- serialization names the result data's encoding, as for inputs.\nALTER TABLE runnerq_results\n    ADD COLUMN IF NOT EXISTS serialization TEXT NOT NULL DEFAULT 'json-v1';\n\n-- Worker pools: one row per live engine instance. Heartbeats let us tell\n-- which pools are still alive for cluster-wide capacity reporting.\nCREATE TABLE IF NOT EXISTS runnerq_worker_pools (\n    pool_id        UUID PRIMARY KEY,\n    queue_name     TEXT NOT NULL,\n    max_workers    INTEGER NOT NULL,\n    activity_types TEXT[],\n    started_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    last_seen_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_worker_pools_queue_alive\n    ON runnerq_worker_pools(queue_name, last_seen_at);\n\n-- Command ledger: applied RunnerQ Cloud commands and their results, so a\n-- repeated command_id replays its result instead of applying twice. Rows are\n-- pruned after a week (commands must be replayable for at least 24 hours).\nCREATE TABLE IF NOT EXISTS runnerq_commands (\n    queue_name  TEXT NOT NULL,\n    command_id  TEXT NOT NULL,\n    fingerprint TEXT NOT NULL,\n    kind        TEXT NOT NULL,\n    result      JSONB NOT NULL,\n    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    PRIMARY KEY (queue_name, command_id)\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_commands_created\n    ON runnerq_commands(created_at);\n\n-- Durable references survive handler replay and are removed with consumer trees.\nCREATE TABLE IF NOT EXISTS runnerq_dependencies (\n queue_name TEXT NOT NULL,\n waiter_activity_id UUID NOT NULL,\n result_id UUID NOT NULL,\n producer_activity_id UUID,\n created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n PRIMARY KEY (queue_name, waiter_activity_id, result_id)\n);\nCREATE INDEX IF NOT EXISTS idx_runnerq_dependencies_result\n ON runnerq_dependencies(queue_name, result_id);\n"
   }
 ];
 /** Built in order after the migrations. */
@@ -56,16 +56,34 @@ export const postgresConcurrentIndexes: readonly PostgresConcurrentIndex[] = [
     "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_query_created\nON runnerq_activities (created_at DESC, id DESC)"
   },
   {
-    "name": "idx_runnerq_query_status_created",
+    "name": "idx_runnerq_query_status",
     "table": "runnerq_activities",
-    "replaces": "",
-    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_query_status_created\nON runnerq_activities (status, created_at DESC, id DESC)"
+    "replaces": "idx_runnerq_query_status_created",
+    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_query_status\nON runnerq_activities (status, created_at DESC, id DESC)\nWHERE status \u003c\u003e 'completed'"
   },
   {
-    "name": "idx_runnerq_query_type_created",
+    "name": "idx_runnerq_processing",
     "table": "runnerq_activities",
-    "replaces": "",
-    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_query_type_created\nON runnerq_activities (activity_type, created_at DESC, id DESC)"
+    "replaces": "idx_runnerq_activities_processing",
+    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_processing\nON runnerq_activities (queue_name)\nWHERE status = 'processing'"
+  },
+  {
+    "name": "idx_runnerq_root_children",
+    "table": "runnerq_activities",
+    "replaces": "idx_runnerq_root_id",
+    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_root_children\nON runnerq_activities (root_activity_id)\nWHERE parent_activity_id IS NOT NULL"
+  },
+  {
+    "name": "idx_runnerq_root_terminal",
+    "table": "runnerq_activities",
+    "replaces": "idx_runnerq_root_terminal_age",
+    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_root_terminal\nON runnerq_activities (queue_name, status, completed_at)\nWHERE parent_activity_id IS NULL\n  AND status IN ('completed', 'failed', 'dead_letter', 'cancelled')"
+  },
+  {
+    "name": "idx_runnerq_results_by_owner",
+    "table": "runnerq_results",
+    "replaces": "idx_runnerq_results_owner",
+    "sql": "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_runnerq_results_by_owner\nON runnerq_results (owner_activity_id)\nWHERE owner_activity_id IS NOT NULL"
   }
 ];
 /** schema/postgres/catalog.json. */
@@ -538,36 +556,11 @@ export const postgresCatalog: Catalog = {
     }
   ],
   "indexes": [
-    {
-      "name": "idx_runnerq_activities_processing",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_activities_processing ON runnerq_activities USING btree (queue_name, lease_deadline_ms) WHERE (status = 'processing'::text)"
-    },
     {
       "name": "idx_runnerq_commands_created",
       "table": "runnerq_commands",
       "definition": "CREATE INDEX idx_runnerq_commands_created ON runnerq_commands USING btree (created_at)"
     },
-    {
-      "name": "idx_runnerq_completed_cron",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_completed_cron ON runnerq_activities USING btree (queue_name, completed_at DESC, created_at DESC) WHERE ((status = ANY (ARRAY['completed'::text, 'failed'::text])) AND ((metadata -\u003e\u003e 'source'::text) = 'cron'::text))"
-    },
-    {
-      "name": "idx_runnerq_completed_non_cron",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_completed_non_cron ON runnerq_activities USING btree (queue_name, completed_at DESC, created_at DESC) WHERE ((status = ANY (ARRAY['completed'::text, 'failed'::text])) AND ((metadata -\u003e\u003e 'source'::text) IS DISTINCT FROM 'cron'::text))"
-    },
-    {
-      "name": "idx_runnerq_dead_letter",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_dead_letter ON runnerq_activities USING btree (queue_name, completed_at DESC) WHERE (status = 'dead_letter'::text)"
-    },
-    {
-      "name": "idx_runnerq_dependencies_producer",
-      "table": "runnerq_dependencies",
-      "definition": "CREATE INDEX idx_runnerq_dependencies_producer ON runnerq_dependencies USING btree (queue_name, producer_activity_id)"
-    },
     {
       "name": "idx_runnerq_dependencies_result",
       "table": "runnerq_dependencies",
@@ -599,44 +592,34 @@ export const postgresCatalog: Catalog = {
       "definition": "CREATE INDEX idx_runnerq_parent_id ON runnerq_activities USING btree (parent_activity_id) WHERE (parent_activity_id IS NOT NULL)"
     },
     {
-      "name": "idx_runnerq_query_created",
+      "name": "idx_runnerq_processing",
       "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_query_created ON runnerq_activities USING btree (created_at DESC, id DESC)"
+      "definition": "CREATE INDEX idx_runnerq_processing ON runnerq_activities USING btree (queue_name) WHERE (status = 'processing'::text)"
     },
     {
-      "name": "idx_runnerq_query_status_created",
+      "name": "idx_runnerq_query_created",
       "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_query_status_created ON runnerq_activities USING btree (status, created_at DESC, id DESC)"
+      "definition": "CREATE INDEX idx_runnerq_query_created ON runnerq_activities USING btree (created_at DESC, id DESC)"
     },
     {
-      "name": "idx_runnerq_query_type_created",
+      "name": "idx_runnerq_query_status",
       "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_query_type_created ON runnerq_activities USING btree (activity_type, created_at DESC, id DESC)"
+      "definition": "CREATE INDEX idx_runnerq_query_status ON runnerq_activities USING btree (status, created_at DESC, id DESC) WHERE (status \u003c\u003e 'completed'::text)"
     },
     {
-      "name": "idx_runnerq_results_owner",
+      "name": "idx_runnerq_results_by_owner",
       "table": "runnerq_results",
-      "definition": "CREATE INDEX idx_runnerq_results_owner ON runnerq_results USING btree (queue_name, owner_activity_id) WHERE (owner_activity_id IS NOT NULL)"
-    },
-    {
-      "name": "idx_runnerq_root_id",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_root_id ON runnerq_activities USING btree (root_activity_id) WHERE (root_activity_id IS NOT NULL)"
-    },
-    {
-      "name": "idx_runnerq_root_only",
-      "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_root_only ON runnerq_activities USING btree (queue_name, created_at DESC) WHERE (parent_activity_id IS NULL)"
+      "definition": "CREATE INDEX idx_runnerq_results_by_owner ON runnerq_results USING btree (owner_activity_id) WHERE (owner_activity_id IS NOT NULL)"
     },
     {
-      "name": "idx_runnerq_root_status",
+      "name": "idx_runnerq_root_children",
       "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_root_status ON runnerq_activities USING btree (queue_name, status) WHERE (parent_activity_id IS NULL)"
+      "definition": "CREATE INDEX idx_runnerq_root_children ON runnerq_activities USING btree (root_activity_id) WHERE (parent_activity_id IS NOT NULL)"
     },
     {
-      "name": "idx_runnerq_root_terminal_age",
+      "name": "idx_runnerq_root_terminal",
       "table": "runnerq_activities",
-      "definition": "CREATE INDEX idx_runnerq_root_terminal_age ON runnerq_activities USING btree (queue_name, status, completed_at) WHERE ((parent_activity_id IS NULL) AND (status = ANY (ARRAY['completed'::text, 'failed'::text, 'dead_letter'::text])))"
+      "definition": "CREATE INDEX idx_runnerq_root_terminal ON runnerq_activities USING btree (queue_name, status, completed_at) WHERE ((parent_activity_id IS NULL) AND (status = ANY (ARRAY['completed'::text, 'failed'::text, 'dead_letter'::text, 'cancelled'::text])))"
     },
     {
       "name": "idx_runnerq_worker_pools_queue_alive",
@@ -650,7 +633,19 @@ export const postgresCatalog: Catalog = {
     ],
     "indexes": [
       "idx_runnerq_dequeue_effective",
-      "idx_runnerq_dequeue_order"
+      "idx_runnerq_dequeue_order",
+      "idx_runnerq_query_status_created",
+      "idx_runnerq_activities_processing",
+      "idx_runnerq_root_id",
+      "idx_runnerq_root_terminal_age",
+      "idx_runnerq_results_owner",
+      "idx_runnerq_completed_non_cron",
+      "idx_runnerq_completed_cron",
+      "idx_runnerq_dead_letter",
+      "idx_runnerq_root_only",
+      "idx_runnerq_root_status",
+      "idx_runnerq_query_type_created",
+      "idx_runnerq_dependencies_producer"
     ]
   }
 };
diff --git a/src/spec.ts b/src/spec.ts
index df98517..dc7b50f 100644
--- a/src/spec.ts
+++ b/src/spec.ts
@@ -1,7 +1,7 @@
 // Code generated by runnerq-spec tools/gen from constants/constants.json. DO NOT EDIT.
 
 /** The runnerq-spec release these constants come from. */
-export const specVersion = "0.6.0";
+export const specVersion = "0.7.0";
 
 /**
  * The Postgres advisory lock every implementation holds while it creates,
diff --git a/src/storage.ts b/src/storage.ts
index c9e4810..cc84c62 100644
--- a/src/storage.ts
+++ b/src/storage.ts
@@ -59,6 +59,12 @@ export interface Park {
 export interface Retention {
   completedMs?: number;
   failedMs?: number;
+  /**
+   * How long the events of finished activities are kept, when that should be shorter than
+   * their tree. 0 (the default) keeps them with the tree; events of unfinished activities
+   * are always kept.
+   */
+  eventsMs?: number;
   intervalMs?: number;
   batchSize?: number;
 }
diff --git a/src/worker.ts b/src/worker.ts
index df9255d..2fd2b38 100644
--- a/src/worker.ts
+++ b/src/worker.ts
@@ -183,6 +183,7 @@ export class Worker
     if (config.retention) {
       integer(config.retention.completedMs ?? 0, "completedMs");
       integer(config.retention.failedMs ?? 0, "failedMs");
+      integer(config.retention.eventsMs ?? 0, "eventsMs");
       integer(config.retention.intervalMs ?? 600_000, "retention interval", 1);
       integer(config.retention.batchSize ?? 100, "retention batch", 1);
     }
diff --git a/test/conductor-queries.test.mjs b/test/conductor-queries.test.mjs
index 0799c0e..4a6b92d 100644
--- a/test/conductor-queries.test.mjs
+++ b/test/conductor-queries.test.mjs
@@ -15,7 +15,8 @@ import { dsn, quiet, setup, submission, until } from "./helpers.mjs";
 
 const integration = (name, fn) =>
   test(name, { skip: !dsn, timeout: 60_000 }, fn);
-const created = "activity.created";
+// Submissions store no events; a signal does.
+const signaled = "signal.received";
 
 /** An agent for a worker on `storage`, connected to a fake gateway. */
 async function connect(t, storage, { config, frame, metadataOnly } = {}) {
@@ -63,6 +64,13 @@ async function enqueue(storage, { type = "Echo", payload = {}, parent } = {}) {
   return a.id;
 }
 
+/** enqueue, then signal the activity, which stores an event. */
+async function enqueueSignaled(storage, options) {
+  const id = await enqueue(storage, options);
+  await storage.signal(id, "go", { serialization: "json-v1", data: {} });
+  return id;
+}
+
 /** Scopes a filter to the test's queue (queries span every queue). */
 function inQueue(queue, ...terms) {
   return { and: [{ field: "queue", op: "eq", value: queue }, ...terms] };
@@ -112,8 +120,11 @@ integration("queries are answered from storage", async (t) => {
   ])
     assert.ok(hello.capabilities[type], `capability ${type}`);
 
-  const id = await enqueue(storage, { payload: { n: 1 } });
-  const child = await enqueue(storage, { payload: { n: 2 }, parent: id });
+  const id = await enqueueSignaled(storage, { payload: { n: 1 } });
+  const child = await enqueueSignaled(storage, {
+    payload: { n: 2 },
+    parent: id,
+  });
   for (let i = 0; i < 3; i++) await enqueue(storage, { type: "Other" });
 
   const echo = inQueue(queue, { field: "type", op: "eq", value: "Echo" });
@@ -143,8 +154,8 @@ integration("queries are answered from storage", async (t) => {
   assert.equal(act.max_attempts, undefined, "unlimited attempts");
   assert.equal(act.timeout_ms, 300_000);
   assert.deepEqual(act.payload, { n: 1 });
-  assert.equal(act.events[0].type, created);
-  assert.deepEqual(act.steps, []);
+  assert.equal(act.events[0].type, signaled);
+  assert.equal(act.steps.length, 1, "the signal's result");
   await g.fails("activities.get", { id: randomUUID() }, "not_found");
   await g.fails("activities.get", { id: "not-an-id" }, "not_found");
 
@@ -303,15 +314,6 @@ integration("a running worker's activities, steps and results", async (t) => {
     }),
     { count: 1, exact: true },
   );
-  const started = await g.ok("events.list", {
-    filter: {
-      and: [
-        { field: "activity_id", op: "eq", value: h.id },
-        { field: "type", op: "eq", value: "attempt.started" },
-      ],
-    },
-  });
-  assert.equal(started.items[0].executor_id, worker.id);
 
   release();
   await h.result({ signal: AbortSignal.timeout(10_000) });
@@ -335,6 +337,16 @@ integration("a running worker's activities, steps and results", async (t) => {
     include: ["last_error", "result"],
   });
   assert.equal(failed.status, "failed");
+  // The failed attempt's event names its executor.
+  const ended = await g.ok("events.list", {
+    filter: {
+      and: [
+        { field: "activity_id", op: "eq", value: refused.id },
+        { field: "type", op: "eq", value: "attempt.failed" },
+      ],
+    },
+  });
+  assert.equal(ended.items[0].executor_id, worker.id);
   assert.equal(failed.last_error.message, "card declined");
   assert.equal(failed.last_error.kind, "non_retryable");
   assert.deepEqual(failed.result, {
@@ -440,9 +452,11 @@ for (const [name, options] of [
       filter: queueFilter(queue),
       max_delay_ms: 50,
     });
-    const later = await enqueue(storage, { payload: { secret: "pii" } });
+    const later = await enqueueSignaled(storage, {
+      payload: { secret: "pii" },
+    });
     const { frames } = await collect(g, sub.subscription_id, [
-      `${later}/${created}`,
+      `${later}/${signaled}`,
     ]);
     assert.doesNotMatch(JSON.stringify(frames.map((f) => f.data)), /detail/);
   });
@@ -465,18 +479,21 @@ integration(
   async (t) => {
     const { storage, queue } = await setup(t);
     const { g } = await connect(t, storage);
-    const before = await enqueue(storage); // before the subscription: not streamed
+    const before = await enqueueSignaled(storage); // before the subscription: not streamed
     const sub = await g.ok("events.subscribe", {
       filter: queueFilter(queue),
       max_delay_ms: 50,
     });
     assert.match(sub.subscription_id, /^sub_[0-9a-f-]{36}$/);
     assert.match(sub.cursor, /^\d+$/);
-    const first = await enqueue(storage);
+    const first = await enqueueSignaled(storage);
     const { seen, cursor } = await collect(g, sub.subscription_id, [
-      `${first}/${created}`,
+      `${first}/${signaled}`,
     ]);
-    assert.ok(!seen.has(`${before}/${created}`), "an older event was streamed");
+    assert.ok(
+      !seen.has(`${before}/${signaled}`),
+      "an older event was streamed",
+    );
     assert.deepEqual(
       await g.ok("events.unsubscribe", {
         subscription_id: sub.subscription_id,
@@ -491,7 +508,7 @@ integration(
 
     // Events while nobody is subscribed reach a subscription resuming after the last
     // cursor; nothing already delivered is repeated.
-    const missed = await enqueue(storage);
+    const missed = await enqueueSignaled(storage);
     const again = await g.ok("events.subscribe", {
       filter: queueFilter(queue),
       after_cursor: cursor,
@@ -499,10 +516,10 @@ integration(
     });
     assert.equal(again.cursor, cursor);
     const resumed = await collect(g, again.subscription_id, [
-      `${missed}/${created}`,
+      `${missed}/${signaled}`,
     ]);
     assert.ok(
-      !resumed.seen.has(`${first}/${created}`),
+      !resumed.seen.has(`${first}/${signaled}`),
       "a resumed stream repeated an event before its cursor",
     );
   },
@@ -524,14 +541,14 @@ integration(
       await client.query("BEGIN");
       const late = randomUUID();
       await client.query(
-        `INSERT INTO runnerq_events (activity_id, queue_name, event_type) VALUES ($1, $2, 'Enqueued')`,
+        `INSERT INTO runnerq_events (activity_id, queue_name, event_type) VALUES ($1, $2, 'Signaled')`,
         [late, queue],
       );
-      const after = await enqueue(storage);
-      await collect(g, sub.subscription_id, [`${after}/${created}`]);
+      const after = await enqueueSignaled(storage);
+      await collect(g, sub.subscription_id, [`${after}/${signaled}`]);
       await client.query("COMMIT");
       const { seen } = await collect(g, sub.subscription_id, [
-        `${late}/${created}`,
+        `${late}/${signaled}`,
       ]);
       for (const [k, n] of seen)
         assert.equal(n, 1, `${k} delivered ${n} times`);
@@ -581,21 +598,21 @@ integration("stream gaps, limits and filters", async (t) => {
     "resource_exhausted",
   );
 
-  // Filters apply to the stream: only this queue's "created" events, read from the start
+  // Filters apply to the stream: only this queue's signal events, read from the start
   // of the log.
   await g.ok("events.unsubscribe", { subscription_id: sub.subscription_id });
-  const id = await enqueue(storage);
+  const id = await enqueueSignaled(storage);
   const filtered = await g.ok("events.subscribe", {
     after_cursor: "0",
     max_delay_ms: 50,
-    filter: inQueue(queue, { field: "type", op: "eq", value: created }),
+    filter: inQueue(queue, { field: "type", op: "eq", value: signaled }),
   });
   assert.equal(filtered.cursor, "0");
   const { frames } = await collect(g, filtered.subscription_id, [
-    `${id}/${created}`,
+    `${id}/${signaled}`,
   ]);
   for (const f of frames)
-    for (const ev of f.data.items) assert.equal(ev.type, created);
+    for (const ev of f.data.items) assert.equal(ev.type, signaled);
 });
 
 // A catch-up batch bigger than the Cloud's frame limit goes out as several frames, each
@@ -616,14 +633,14 @@ integration("a stream splits batches to the frame limit", async (t) => {
     want.set(id, size);
     await pool.query(
       `INSERT INTO runnerq_events (activity_id, queue_name, event_type, detail)
-      VALUES ($1, $2, 'Enqueued', jsonb_build_object('blob', repeat('x', $3::int)))`,
+      VALUES ($1, $2, 'Signaled', jsonb_build_object('blob', repeat('x', $3::int)))`,
       [id, queue, size],
     );
   }
   const { frames } = await collect(
     g,
     sub.subscription_id,
-    [...want.keys()].map((id) => `${id}/${created}`),
+    [...want.keys()].map((id) => `${id}/${signaled}`),
   );
   let last = 0n;
   const count = new Map();
@@ -661,7 +678,7 @@ integration(
     for (let i = 0; i < 3; i++)
       await pool.query(
         `INSERT INTO runnerq_events (activity_id, queue_name, event_type, detail)
-      VALUES ($1, $2, 'Enqueued', jsonb_build_object('blob', repeat('x', 1000000)))`,
+      VALUES ($1, $2, 'Signaled', jsonb_build_object('blob', repeat('x', 1000000)))`,
         [randomUUID(), queue],
       );
     await g.until(
@@ -694,9 +711,9 @@ integration(
           max_delay_ms: 50,
         }),
       );
-    const id = await enqueue(storage);
+    const id = await enqueueSignaled(storage);
     for (const s of subs)
-      await collect(g, s.subscription_id, [`${id}/${created}`]);
+      await collect(g, s.subscription_id, [`${id}/${signaled}`]);
     assert.ok(
       !g.events
         .slice(mark)
diff --git a/test/conformance-driver.mjs b/test/conformance-driver.mjs
index 8177762..8882acb 100644
--- a/test/conformance-driver.mjs
+++ b/test/conformance-driver.mjs
@@ -112,6 +112,7 @@ const ops = {
       count: await storage.cleanup({
         completedMs: (a.completed_s ?? 0) * 1000,
         failedMs: (a.failed_s ?? 0) * 1000,
+        eventsMs: (a.events_s ?? 0) * 1000,
         batchSize: a.batch ?? 1,
       }),
     };
diff --git a/test/postgres.test.mjs b/test/postgres.test.mjs
index 28f0382..7e877fd 100644
--- a/test/postgres.test.mjs
+++ b/test/postgres.test.mjs
@@ -165,10 +165,7 @@ integration(
         code: "checkpoint_conflict",
       },
     );
-    assert.equal(
-      (await storage.events(a.id)).filter((e) => e.type === "Completed").length,
-      1,
-    );
+    assert.deepEqual(await storage.events(a.id), [], "success stores no event");
     const b = submission(),
       bf = await claim(storage, b);
     assert.equal(await storage.fail(bf, "oops", true), "retrying");
diff --git a/test/query-compiler.test.mjs b/test/query-compiler.test.mjs
index e9d7666..c2fed1b 100644
--- a/test/query-compiler.test.mjs
+++ b/test/query-compiler.test.mjs
@@ -58,12 +58,11 @@ test("filters compile to parameterised SQL", () => {
   });
   assert.equal(
     sql,
-    "(a.status = ANY($1::text[]) AND a.activity_type = ANY($2::text[]) AND a.created_at >= $3::timestamptz" +
-      " AND (a.metadata->>$4::text) = ANY($5::text[]) AND (NOT a.parent_activity_id IS NOT NULL)" +
-      " AND (a.priority > $6::bigint OR left(a.queue_name, char_length($7::text)) = $7::text))",
+    "(a.status IN ('scheduled', 'retrying', 'processing') AND a.activity_type = ANY($1::text[]) AND a.created_at >= $2::timestamptz" +
+      " AND (a.metadata->>$3::text) = ANY($4::text[]) AND (NOT a.parent_activity_id IS NOT NULL)" +
+      " AND (a.priority > $5::bigint OR left(a.queue_name, char_length($6::text)) = $6::text))",
   );
   assert.deepEqual(args, [
-    ["scheduled", "retrying", "processing"],
     ["charge_card"],
     "2026-09-01T00:00:00Z",
     "tenant",
@@ -232,21 +231,16 @@ test("event filters: canonical types, roots and seq", () => {
   assert.equal(
     sql,
     "(e.event_type = ANY($1::text[]) AND e.activity_id IN (SELECT x.id FROM runnerq_activities x" +
-      " WHERE x.id = ANY($2::uuid[]) OR x.root_activity_id = ANY($2::uuid[])) AND e.id > $3::bigint)",
+      " WHERE x.id = ANY($2::uuid[]) OR (x.root_activity_id = ANY($2::uuid[]) AND x.parent_activity_id IS NOT NULL)) AND e.id > $3::bigint)",
   );
   assert.deepEqual(args, [["Failed", "Retrying", "custom"], [id], "10"]);
 });
 
 test("every event the TypeScript storage writes has a canonical type", () => {
   const written = [
-    "Enqueued",
-    "Scheduled",
-    "Dequeued",
-    "Completed",
     "Retrying",
     "DeadLetter",
     "Failed",
-    "ResultStored",
     "Yielded",
     "Signaled",
     "SpawnLinked",
@@ -264,8 +258,6 @@ test("every event the TypeScript storage writes has a canonical type", () => {
     assert.ok(!canonical.startsWith("other."), `${name} has no canonical type`);
     assert.ok(internalEvents(canonical).includes(name));
   }
-  assert.equal(canonicalEvent("Enqueued"), "activity.created");
-  assert.equal(canonicalEvent("Dequeued"), "attempt.started");
   assert.equal(canonicalEvent("Requeued"), "attempt.lease_expired");
   assert.equal(canonicalEvent("Brand New"), "other.brand new");
   assert.equal(canonicalEvent("constructor"), "other.constructor");
diff --git a/test/schema.test.mjs b/test/schema.test.mjs
index 4d3fcc5..b724ece 100644
--- a/test/schema.test.mjs
+++ b/test/schema.test.mjs
@@ -96,14 +96,21 @@ integration(
   async (t) => {
     const { connectionString, pool } = await freshSchema(t);
     await PostgresStorage.initialize({ connectionString });
-    // What older versions of this SDK created: no ledger, no query indexes.
+    // What older versions of this SDK created: no ledger, none of the newer indexes, and
+    // an index since retired.
     await pool.query("DROP TABLE runnerq_commands");
     for (const name of [
       "idx_runnerq_query_created",
-      "idx_runnerq_query_status_created",
-      "idx_runnerq_query_type_created",
+      "idx_runnerq_query_status",
+      "idx_runnerq_processing",
+      "idx_runnerq_root_children",
+      "idx_runnerq_root_terminal",
+      "idx_runnerq_results_by_owner",
     ])
       await pool.query(`DROP INDEX ${name}`);
+    await pool.query(
+      "CREATE INDEX idx_runnerq_root_status ON runnerq_activities(queue_name, status) WHERE parent_activity_id IS NULL",
+    );
     const storage = await PostgresStorage.connect({
       connectionString,
       queue: "older",