From be8b9540ccd197773d626a7ba945ece531d4bcea Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:08:04 -0700
Subject: [PATCH 1/7] fix(agents): bound the assigned-task wake per row, and
 end a run whose post the run cap refuses

A row an agent created for itself now wakes it once, then again only after a
person acts in that pod. Rows a person created or assigned stop after three
consecutive failed runs, with one note on the row; a person's update or a
successful run resets the count. No wake while the agent's own posts are
still unanswered. Per-row state lives in AgentTaskRowWakeState.

A commonly_post_message refused by consecutive_run_cap now ends the hosted
run that round, settled as no_action with that reason, instead of retrying
the post for the rest of its turns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.test.js      | 183 +++++++++++++++++
 .../agentEventService.lifecycle.test.js       |  13 ++
 .../nativeRuntimeService.webSearch.test.js    |  33 ++++
 backend/models/AgentTaskRowWakeState.ts       |  29 +++
 backend/services/agentEventService.ts         |  13 +-
 backend/services/agentMessageService.ts       |  11 ++
 backend/services/assignedTaskWakeService.ts   | 184 +++++++++++++++++-
 backend/services/nativeRuntimeService.ts      |  29 +++
 8 files changed, 487 insertions(+), 8 deletions(-)
 create mode 100644 backend/__tests__/service/assignedTaskWakeRows.test.js
 create mode 100644 backend/models/AgentTaskRowWakeState.ts

diff --git a/backend/__tests__/service/assignedTaskWakeRows.test.js b/backend/__tests__/service/assignedTaskWakeRows.test.js
new file mode 100644
index 0000000000..11bc5edd60
--- /dev/null
+++ b/backend/__tests__/service/assignedTaskWakeRows.test.js
@@ -0,0 +1,183 @@
+const mongoose = require('mongoose');
+const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils');
+jest.mock('jsonwebtoken', () => ({}));
+jest.mock('../../services/agentEventService', () => ({ enqueue: jest.fn() }));
+const Task = require('../../models/Task');
+const User = require('../../models/User');
+const Message = require('../../models/Message');
+const AgentEvent = require('../../models/AgentEvent');
+const { AgentInstallation } = require('../../models/AgentRegistry');
+const AgentEventService = require('../../services/agentEventService');
+const { dispatchAssignedTaskWakes } = require('../../services/assignedTaskWakeService');
+
+const STOPPED = 'I stopped retrying this after repeated errors. Reassign it or update it to try again.';
+const NOW = new Date('2026-10-24T11:00:00Z');
+const at = (minutes) => new Date(NOW.getTime() + minutes * 60000);
+let podId;
+let agent;
+let person;
+
+const eventRow = (extra) => ({
+  agentName: 'codex', instanceId: 'forge', podId, type: 'chat.mention', deliveryMode: 'wake', ...extra,
+});
+
+// How the run a wake started ended, as the queue records it.
+async function settle(outcome, minutes) {
+  const { _id } = await AgentEventService.enqueue.mock.results.at(-1).value;
+  const when = at(minutes);
+  const delivery = outcome === 'failed'
+    ? { outcome: 'error', updatedAt: when }
+    : { outcome: 'no_action', reasonRecorded: true, updatedAt: when };
+  await AgentEvent.collection.insertOne(eventRow({
+    _id, status: outcome === 'failed' ? 'failed' : 'acked', delivery, createdAt: when, updatedAt: when,
+  }));
+}
+
+const wakes = async (minutes) => (await dispatchAssignedTaskWakes(at(minutes))).woken;
+const say = (userId, minutes) => Message.collection.insertOne({
+  podId, userId, content: 'hello', messageType: 'text', createdAt: at(minutes),
+});
+const row = () => Task.findOne({ podId }).lean();
+
+beforeAll(setupMongoDb);
+afterAll(closeMongoDb);
+beforeEach(async () => {
+  await clearMongoDb();
+  podId = new mongoose.Types.ObjectId();
+  agent = await User.create({
+    username: 'forge-row-test', email: 'forge-row@example.test', password: 'test-password',
+    isBot: true, botType: 'agent', botMetadata: { agentName: 'codex', instanceId: 'forge' },
+  });
+  person = await User.create({ username: 'row-owner', email: 'row-owner@example.test', password: 'test-password' });
+  await AgentInstallation.create({
+    agentName: 'codex', instanceId: 'forge', podId, version: '1', installedBy: agent._id,
+    config: { boardWake: { enabled: false } }, status: 'active',
+  });
+  const when = at(-6);
+  await AgentEvent.collection.insertOne(eventRow({
+    status: 'acked', delivery: { outcome: 'no_action', reasonRecorded: true, updatedAt: when }, createdAt: when, updatedAt: when,
+  }));
+  AgentEventService.enqueue.mockReset().mockImplementation(async () => ({ _id: new mongoose.Types.ObjectId() }));
+});
+
+describe('a row the agent created for itself', () => {
+  beforeEach(() => Task.create({
+    podId, taskNum: 4, taskId: 'N-4-codex-mvid77qk', title: 'Keep posting updates', assignee: 'forge', source: 'agent',
+  }));
+
+  it('wakes once, then stays quiet while nobody acts', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    expect(await wakes(31)).toBe(0);
+    expect(await wakes(240)).toBe(0);
+  });
+
+  it('stays quiet after its own run fails, too', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('failed', 1);
+    expect(await wakes(31)).toBe(0);
+  });
+
+  it('is not re-armed by the agent itself speaking', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    await say(agent._id, 10);
+    expect(await wakes(31)).toBe(0);
+  });
+
+  it('is re-armed for exactly one wake by a person posting in the pod', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    await say(person._id, 10);
+    expect(await wakes(31)).toBe(1);
+    await settle('succeeded', 32);
+    expect(await wakes(62)).toBe(0);
+  });
+
+  it('is re-armed by a person creating or updating a row in the pod', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    await Task.create({
+      podId, taskNum: 5, taskId: 'TASK-5', title: 'Unrelated', status: 'done',
+      updates: [{ text: 'Created by row-owner', author: 'row-owner', authorId: String(person._id), createdAt: at(10) }],
+    });
+    expect(await wakes(31)).toBe(1);
+  });
+
+  it('is recognised from its creator update when the id carries no agent name', async () => {
+    await Task.updateOne({ podId }, { $set: {
+      taskId: 'TASK-4',
+      updates: [{ text: 'Created by forge', author: 'forge', authorId: String(agent._id), createdAt: at(-60) }],
+    } });
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    expect(await wakes(31)).toBe(0);
+  });
+});
+
+describe('a row a person created or assigned', () => {
+  beforeEach(() => Task.create({ podId, taskNum: 1, taskId: 'TASK-1', title: 'Do the assigned work', assignee: 'forge' }));
+
+  it('keeps waking on the normal cadence while runs succeed', async () => {
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    expect(await wakes(31)).toBe(1);
+    await settle('succeeded', 32);
+    expect(await wakes(62)).toBe(1);
+  });
+
+  it('stops after three failed runs in a row, and says so once on the row', async () => {
+    const before = (await row()).updatedAt;
+    for (const minutes of [0, 31, 62]) {
+      expect(await wakes(minutes)).toBe(1); // eslint-disable-line no-await-in-loop
+      await settle('failed', minutes + 1); // eslint-disable-line no-await-in-loop
+    }
+    expect(await wakes(93)).toBe(0);
+    expect(await wakes(124)).toBe(0);
+    const stopped = await row();
+    expect(stopped.updates.filter((update) => update.text === STOPPED)).toEqual([
+      expect.objectContaining({ author: 'codex', authorId: String(agent._id) }),
+    ]);
+    expect(stopped.updatedAt).toEqual(before);
+  });
+
+  it('starts again after a person updates the stopped row', async () => {
+    for (const minutes of [0, 31, 62]) {
+      await wakes(minutes); // eslint-disable-line no-await-in-loop
+      await settle('failed', minutes + 1); // eslint-disable-line no-await-in-loop
+    }
+    expect(await wakes(93)).toBe(0);
+    await Task.updateOne({ podId }, { $push: { updates: {
+      text: 'Try with the new key', author: 'row-owner', authorId: String(person._id), createdAt: at(100),
+    } } });
+    expect(await wakes(124)).toBe(1);
+    await settle('failed', 125);
+    expect(await wakes(155)).toBe(1);
+  });
+
+  it('is not restarted by a pod message alone', async () => {
+    for (const minutes of [0, 31, 62]) {
+      await wakes(minutes); // eslint-disable-line no-await-in-loop
+      await settle('failed', minutes + 1); // eslint-disable-line no-await-in-loop
+    }
+    await say(person._id, 70);
+    expect(await wakes(93)).toBe(0);
+  });
+
+  it('counts only consecutive failures: a successful run resets the count', async () => {
+    const outcomes = ['failed', 'failed', 'succeeded', 'failed', 'failed'];
+    for (const [step, outcome] of outcomes.entries()) {
+      expect(await wakes(step * 31)).toBe(1); // eslint-disable-line no-await-in-loop
+      await settle(outcome, step * 31 + 1); // eslint-disable-line no-await-in-loop
+    }
+    expect(await wakes(outcomes.length * 31)).toBe(1);
+  });
+});
+
+it('does not wake an agent whose last posts in the pod are still unanswered', async () => {
+  await Task.create({ podId, taskNum: 1, taskId: 'TASK-1', title: 'Do the assigned work', assignee: 'forge' });
+  for (const minutes of [-30, -20, -10]) await say(agent._id, minutes); // eslint-disable-line no-await-in-loop
+  expect(await wakes(0)).toBe(0);
+  await say(person._id, 1);
+  expect(await wakes(2)).toBe(1);
+});
diff --git a/backend/__tests__/unit/services/agentEventService.lifecycle.test.js b/backend/__tests__/unit/services/agentEventService.lifecycle.test.js
index 96699066e9..97c386ba9e 100644
--- a/backend/__tests__/unit/services/agentEventService.lifecycle.test.js
+++ b/backend/__tests__/unit/services/agentEventService.lifecycle.test.js
@@ -610,6 +610,19 @@ describe('driver paths reach a terminal state', () => {
       acknowledge.mockRestore();
     });
 
+    test('a run that stopped because the agent holds the floor settles as a decline with that reason', async () => {
+      AgentEvent.find.mockReturnValue({ sort: () => ({ limit: () => ({ lean: () => Promise.resolve([]) }) }) });
+      runAgent.mockResolvedValue({ status: 'succeeded', runId: 'run-1', totalTurns: 2, noActionReason: 'consecutive_run_cap' });
+      const acknowledge = jest.spyOn(AgentEventService, 'acknowledge').mockResolvedValue({});
+      await enqueueNative();
+      expect(acknowledge).toHaveBeenCalledWith(
+        'evt-n', 'clerk', 'default',
+        { outcome: 'no_action', reason: 'consecutive_run_cap' },
+        null, 'native',
+      );
+      acknowledge.mockRestore();
+    });
+
     test('a podId shaped like a query operator is never looked up or run in-process', async () => {
       AgentInstallation.findOne.mockReturnValue({ lean: () => Promise.resolve(nativeInstall) });
       AgentEvent.create.mockResolvedValue({ _id: 'evt-x', agentName: 'clerk', instanceId: 'default', type: 'chat.mention', payload: {} });
diff --git a/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js b/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
index 67e778cdc3..e02a96fe13 100644
--- a/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
+++ b/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
@@ -139,6 +139,39 @@ describe('native turn with commonly_web_search', () => {
     expect(JSON.parse(toolMessages(1)[0].content)).toEqual(expect.objectContaining({ posted, ok: posted }));
   });
 
+  test('a post refused because the agent holds the floor ends the run that round, as a decline', async () => {
+    const capped = { success: false, refused: true, reason: 'consecutive_run_cap', consecutive: 3, guidance: 'attach it with commonly_attach_file' };
+    AgentMessageService.postMessage.mockResolvedValue(capped);
+    const post = (id) => toolCalls(call(id, 'commonly_post_message', { content: 'Here is the update.' }));
+    axios.post
+      .mockResolvedValueOnce(toolCalls(call('s1', 'commonly_web_search', { query: 'q' })))
+      .mockResolvedValueOnce(post('p1'))
+      .mockResolvedValueOnce(post('p2'))
+      .mockResolvedValueOnce(post('p3'))
+      .mockResolvedValueOnce(finalReply);
+
+    const result = await runAgent(install, mention);
+
+    expect(axios.post).toHaveBeenCalledTimes(2);
+    expect(AgentMessageService.postMessage).toHaveBeenCalledTimes(1);
+    expect(result).toEqual(expect.objectContaining({ status: 'succeeded', totalTurns: 2, noActionReason: 'consecutive_run_cap' }));
+    expect(savedRun.turns[1].toolCalls[0]).toEqual(expect.objectContaining({ error: 'message_post_refused' }));
+    const refusal = JSON.stringify(savedRun.turns[1].toolCalls[0].result);
+    expect(refusal).toContain('Wait for someone else to speak');
+    expect(refusal).not.toContain('commonly_attach_file');
+    // The queued retries were never requested; drop them so they cannot answer the next test.
+    axios.post.mockReset();
+    AgentMessageService.postMessage.mockReset().mockResolvedValue({ success: true });
+  });
+
+  test('a run that posts after an earlier capped refusal reports no decline', async () => {
+    AgentMessageService.postMessage.mockResolvedValueOnce({ success: true, message: { id: 'posted-1' } });
+    axios.post.mockResolvedValueOnce(toolCalls(call('p1', 'commonly_post_message', { content: 'Done.' })))
+      .mockResolvedValueOnce(finalReply);
+    const result = await runAgent(install, mention);
+    expect(result.noActionReason).toBeUndefined();
+  });
+
   test('results reach the model marked untrusted; the run keeps the count and the addresses only', async () => {
     axios.post
       .mockResolvedValueOnce(toolCalls(call('c1', 'commonly_web_search', { query: 'node 26 release' })))
diff --git a/backend/models/AgentTaskRowWakeState.ts b/backend/models/AgentTaskRowWakeState.ts
new file mode 100644
index 0000000000..8e1e29fced
--- /dev/null
+++ b/backend/models/AgentTaskRowWakeState.ts
@@ -0,0 +1,29 @@
+import mongoose, { Document, Schema, Types } from 'mongoose';
+
+interface IAgentTaskRowWakeState extends Document {
+  _id: string;
+  lastWakeAt: Date | null;
+  lastWakeEventId: Types.ObjectId | null;
+  // The wake event whose outcome is already in `failures`, so one run counts once.
+  countedEventId: Types.ObjectId | null;
+  failures: number;
+  stoppedAt: Date | null;
+}
+
+// What the assigned-task wake remembers about one row for one seat, keyed
+// `<seat key>|<row _id>`. AgentTaskWakeState throttles the seat; this decides
+// whether a row may wake it at all: a row the agent wrote for itself wakes it
+// once per human action, and any row stops after repeated failed runs.
+const schema = new Schema<IAgentTaskRowWakeState>({
+  _id: { type: String, required: true },
+  lastWakeAt: { type: Date, default: null },
+  lastWakeEventId: { type: Schema.Types.ObjectId, default: null },
+  countedEventId: { type: Schema.Types.ObjectId, default: null },
+  failures: { type: Number, default: 0 },
+  stoppedAt: { type: Date, default: null },
+}, { timestamps: true });
+
+const AgentTaskRowWakeState = mongoose.model<IAgentTaskRowWakeState>('AgentTaskRowWakeState', schema);
+export default AgentTaskRowWakeState;
+// eslint-disable-next-line @typescript-eslint/no-require-imports
+module.exports = AgentTaskRowWakeState;
diff --git a/backend/services/agentEventService.ts b/backend/services/agentEventService.ts
index ef5fbaef44..e338d00f17 100644
--- a/backend/services/agentEventService.ts
+++ b/backend/services/agentEventService.ts
@@ -1664,7 +1664,9 @@ class AgentEventService {
       // actually succeeded as 'failed'. The rejection handler here is scoped
       // to runAgent alone. Both terminal calls swallow their own errors so
       // this stays fire-and-forget for the caller.
-        (runResult: { status?: string; error?: string; runId?: string; totalTurns?: number }) => {
+        (runResult: {
+          status?: string; error?: string; runId?: string; totalTurns?: number; noActionReason?: string;
+        }) => {
           turnFinished();
           if (runResult?.status === 'drained') return undefined;
           return (runResult?.status === 'failed'
@@ -1679,9 +1681,12 @@ class AgentEventService {
           eventIdStr,
           agentName,
           instanceId,
-          runResult?.runId && Number(runResult.totalTurns) > 0
-            ? { outcome: 'acknowledged', reason: 'native-runtime-completed' }
-            : { outcome: 'no_action', reason: 'native-no-turn' },
+          // eslint-disable-next-line no-nested-ternary
+          runResult?.noActionReason
+            ? { outcome: 'no_action', reason: runResult.noActionReason }
+            : runResult?.runId && Number(runResult.totalTurns) > 0
+              ? { outcome: 'acknowledged', reason: 'native-runtime-completed' }
+              : { outcome: 'no_action', reason: 'native-no-turn' },
           nativeDeliveryId,
           'native',
         ).catch((ackErr: Error) => {
diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts
index 1fa51684d1..1019f04858 100644
--- a/backend/services/agentMessageService.ts
+++ b/backend/services/agentMessageService.ts
@@ -843,6 +843,17 @@ class AgentMessageService {
     return run;
   }
 
+  /**
+   * Whether this agent's next post in the pod would be refused by the run cap.
+   * The same pieces postMessage reads: no cap in a 1:1 or an agent-admin pod.
+   */
+  static async holdsFloor(podId: unknown, userId: unknown, podType: unknown): Promise<boolean> {
+    if (podType === 'agent-admin' || AgentMessageService.isOneToOnePod(podType)) return false;
+    const cap = AgentMessageService.resolveConsecutiveRunCap();
+    if (cap <= 0) return false;
+    return (await AgentMessageService.countConsecutiveRun(podId, userId)) >= cap;
+  }
+
   static async findRecentDuplicate(options: {
     podId: unknown;
     userId: unknown;
diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts
index 2dd15bd57c..c95d9a32cf 100644
--- a/backend/services/assignedTaskWakeService.ts
+++ b/backend/services/assignedTaskWakeService.ts
@@ -1,12 +1,18 @@
+import { Types } from 'mongoose';
 import AgentTaskWakeState from '../models/AgentTaskWakeState';
+import AgentTaskRowWakeState from '../models/AgentTaskRowWakeState';
 import AgentEvent from '../models/AgentEvent';
 import AgentRun from '../models/AgentRun';
 import Task from '../models/Task';
 import User from '../models/User';
+import Pod from '../models/Pod';
 import { AgentInstallation } from '../models/AgentRegistry';
 
 const DELAY_MS = 5 * 60 * 1000;
 const INTERVAL_MS = 30 * 60 * 1000;
+// Consecutive failed wake runs on one row before the wake gives up on it.
+const STOP_AFTER_FAILURES = 3;
+const STOPPED_NOTE = 'I stopped retrying this after repeated errors. Reassign it or update it to try again.';
 const seatKey = (name: unknown, instance: unknown): string => JSON.stringify([
   String(name || '').toLowerCase(), String(instance || 'default').toLowerCase(),
 ]);
@@ -19,7 +25,158 @@ type Installation = { agentName: string; instanceId: string; podId: unknown };
 type Row = {
   _id: unknown; podId: unknown; taskId: string; title: string; status: string;
   assignee?: string | null; claimedBy?: string | null; dep?: string | null; depMockOk?: boolean;
+  source?: string; updates?: Array<{ authorId?: string | null; createdAt?: Date }>;
 };
+type RowState = {
+  _id: string; lastWakeAt?: Date | null; lastWakeEventId?: unknown; countedEventId?: unknown;
+  failures?: number; stoppedAt?: Date | null;
+};
+type WakeEvent = { _id: unknown; status?: string; delivery?: { outcome?: string; reason?: string | null } };
+
+const rowKey = (seat: string, row: Row): string => `${seat}|${String(row._id)}`;
+// Native rows record no author: their id is minted `N-<num>-<agentName>-<base36>`.
+// Rows created through the API open with a "Created by" update carrying the author's id.
+const NATIVE_TASK_ID = /^N-\d+-(.+)-[0-9a-z]+$/;
+
+function selfCreated(row: Row, install: Installation, principalId: string): boolean {
+  if (row.source !== 'agent') return false;
+  const authorId = row.updates?.[0]?.authorId;
+  if (authorId) return String(authorId) === principalId;
+  const minted = NATIVE_TASK_ID.exec(row.taskId);
+  return Boolean(minted) && normalized(minted?.[1]) === normalized(install.agentName);
+}
+
+async function anyHuman(ids: unknown[]): Promise<boolean> {
+  const unique = [...new Set(ids.map((id) => String(id || '')))].filter((id) => Types.ObjectId.isValid(id));
+  if (!unique.length) return false;
+  return Boolean(await User.exists({ _id: { $in: unique }, isBot: { $ne: true } }));
+}
+
+const updatedSince = (row: Row, since: Date): unknown[] => (row.updates || [])
+  .filter((update) => update.createdAt && new Date(update.createdAt) > since)
+  .map((update) => update.authorId);
+
+// A person posted in the pod, or created or updated any row in it, after `since`.
+async function humanActedInPodSince(podId: unknown, since: Date): Promise<boolean> {
+  // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
+  const AgentMessageService = require('./agentMessageService');
+  let authors: unknown[] = [];
+  try {
+    const recent = await AgentMessageService.getRecentMessages(podId, 50) as Array<{
+      createdAt?: unknown; userId?: { _id?: unknown } | unknown; user_id?: unknown;
+    }>;
+    authors = recent.filter((message) => new Date(String(message.createdAt || 0)) > since)
+      .map((message) => (message.userId as { _id?: unknown })?._id || message.userId || message.user_id);
+  } catch (error) {
+    // Unread means no wake: a row the agent set itself waits for a person anyway.
+    console.warn('[assigned-task-wake] pod activity unreadable:', String(podId), (error as Error).message);
+    return false;
+  }
+  const rows = await Task.find({ podId, 'updates.createdAt': { $gt: since } })
+    .select('updates.authorId updates.createdAt').lean() as Row[];
+  return anyHuman([...authors, ...rows.flatMap((row) => updatedSince(row, since))]);
+}
+
+// 'open' until the wake's run settles; a run that did nothing for its own
+// reason is neither a success nor a failure.
+function verdictOf(event: WakeEvent | undefined): 'open' | 'failed' | 'succeeded' | 'neutral' {
+  if (!event) return 'neutral';
+  if (event.status === 'failed' || event.delivery?.outcome === 'error') return 'failed';
+  if (event.status !== 'acked') return ['pending', 'folding', 'delivered'].includes(String(event.status)) ? 'open' : 'neutral';
+  const outcome = event.delivery?.outcome;
+  if (outcome === 'posted' || outcome === 'acknowledged') return 'succeeded';
+  return outcome === 'no_action' && !event.delivery?.reason ? 'succeeded' : 'neutral';
+}
+
+async function countOutcomes(states: RowState[]): Promise<void> {
+  const unsettled = states.filter((state) => state.lastWakeEventId
+    && String(state.lastWakeEventId) !== String(state.countedEventId));
+  if (!unsettled.length) return;
+  const events = await AgentEvent.find({ _id: { $in: unsettled.map((state) => state.lastWakeEventId) } })
+    .select('status delivery.outcome delivery.reason').lean() as WakeEvent[];
+  const byId = new Map(events.map((event) => [String(event._id), event]));
+  for (const state of unsettled) {
+    const verdict = verdictOf(byId.get(String(state.lastWakeEventId)));
+    if (verdict === 'open') continue;
+    const failures = verdict === 'failed' ? (state.failures || 0) + 1 : verdict === 'succeeded' ? 0 : (state.failures || 0);
+    // The prior countedEventId guards against two schedulers counting one run twice.
+    // eslint-disable-next-line no-await-in-loop
+    const counted = await AgentTaskRowWakeState.updateOne(
+      { _id: state._id, countedEventId: state.countedEventId ?? null },
+      { $set: { countedEventId: state.lastWakeEventId, failures } },
+    );
+    if (counted.modifiedCount) {
+      state.countedEventId = state.lastWakeEventId;
+      state.failures = failures;
+    }
+  }
+}
+
+// Says so once on the row, in the agent's name, without touching updatedAt:
+// the note must not read as a fresh revision that offers the row again.
+async function stopRow(row: Row, state: RowState, install: Installation, principalId: string, now: Date): Promise<void> {
+  const stopped = await AgentTaskRowWakeState.updateOne(
+    { _id: state._id, stoppedAt: null },
+    { $set: { stoppedAt: now } },
+  );
+  if (!stopped.modifiedCount) return;
+  state.stoppedAt = now;
+  await Task.updateOne(
+    { _id: row._id },
+    { $push: { updates: { text: STOPPED_NOTE, author: install.agentName, authorId: principalId, createdAt: now } } },
+    { timestamps: false },
+  );
+}
+
+// The rows that may wake this seat now. A row the agent created for itself
+// wakes it once, then again only after a person acts in the pod. Any other
+// row wakes it until STOP_AFTER_FAILURES consecutive runs fail; a person
+// updating the row, or a run that succeeds, starts the count again.
+async function wakeableRows(
+  owned: Row[], install: Installation, principalId: string, seat: string, now: Date,
+): Promise<Row[]> {
+  const states = await AgentTaskRowWakeState.find({ _id: { $in: owned.map((row) => rowKey(seat, row)) } })
+    .lean() as RowState[];
+  await countOutcomes(states);
+  const byRow = new Map(states.map((state) => [state._id, state]));
+  const wakeable: Row[] = [];
+  for (const row of owned) {
+    const state = byRow.get(rowKey(seat, row));
+    if (selfCreated(row, install, principalId)) {
+      // eslint-disable-next-line no-await-in-loop
+      if (!state?.lastWakeAt || await humanActedInPodSince(row.podId, new Date(state.lastWakeAt))) wakeable.push(row);
+      continue;
+    }
+    if (!state) {
+      wakeable.push(row);
+      continue;
+    }
+    const failing = Boolean(state.stoppedAt) || (state.failures || 0) > 0;
+    // eslint-disable-next-line no-await-in-loop
+    if (failing && state.lastWakeAt && await anyHuman(updatedSince(row, new Date(state.lastWakeAt)))) {
+      // eslint-disable-next-line no-await-in-loop
+      await AgentTaskRowWakeState.updateOne({ _id: state._id }, { $set: { failures: 0, stoppedAt: null } });
+      wakeable.push(row);
+      continue;
+    }
+    if (state.stoppedAt) continue;
+    if ((state.failures || 0) >= STOP_AFTER_FAILURES) {
+      // eslint-disable-next-line no-await-in-loop
+      await stopRow(row, state, install, principalId, now);
+      continue;
+    }
+    wakeable.push(row);
+  }
+  return wakeable;
+}
+
+// A post here would be refused until someone else speaks, so a wake could only spend tokens.
+async function holdsFloor(install: Installation, principalId: string): Promise<boolean> {
+  // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
+  const AgentMessageService = require('./agentMessageService');
+  const pod = await Pod.findById(install.podId).select('type').lean() as { type?: string } | null;
+  return AgentMessageService.holdsFloor(install.podId, principalId, pod?.type);
+}
 
 function owns(row: Row, install: Installation, principalId: string): boolean {
   const name = normalized(install.agentName);
@@ -62,7 +219,8 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
       { status: 'claimed', claimedBy: { $type: 'string', $ne: '' } },
     ],
   })
-    .select('_id podId taskId title status assignee claimedBy dep depMockOk').lean() as Row[];
+    .select('_id podId taskId title status assignee claimedBy dep depMockOk source updates.authorId updates.createdAt')
+    .lean() as Row[];
   const rows = await actionableRows(open);
   if (!rows.length) return { eligible: 0, woken: 0 };
   const installs = await AgentInstallation.find({
@@ -83,10 +241,13 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
     if (visited.has(key)) continue;
     const principalId = principalIds.get(key);
     if (!principalId) continue;
-    const owned = rows.filter((row) => String(row.podId) === String(install.podId)
+    const assigned = rows.filter((row) => String(row.podId) === String(install.podId)
       && owns(row, install, principalId));
-    if (!owned.length) continue;
+    if (!assigned.length) continue;
     visited.add(key);
+    // eslint-disable-next-line no-await-in-loop
+    const owned = await wakeableRows(assigned, install, principalId, key, now);
+    if (!owned.length) continue;
     // Historical webhook rows lost their reasons. Silent completions require
     // durable writer provenance, independent of mutable runtime configuration.
     // Skips are not turns and must not hide an earlier completed turn. Use
@@ -111,6 +272,8 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
     if (dueAt > now) continue;
     // eslint-disable-next-line no-await-in-loop
     if (await hasOtherWork(install.agentName, install.instanceId)) continue;
+    // eslint-disable-next-line no-await-in-loop
+    if (await holdsFloor(install, principalId)) continue;
     eligible += 1;
     let reserved = false;
     let enqueueStarted = false;
@@ -143,7 +306,7 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
       const AgentEventService = require('./agentEventService');
       enqueueStarted = true;
       // eslint-disable-next-line no-await-in-loop
-      await AgentEventService.enqueue({
+      const wake = await AgentEventService.enqueue({
         agentName: install.agentName, instanceId: install.instanceId, podId: String(install.podId),
         type: 'chat.mention',
         payload: {
@@ -156,6 +319,19 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
         },
       });
       woken += 1;
+      try {
+        // eslint-disable-next-line no-await-in-loop
+        await AgentTaskRowWakeState.bulkWrite(current.slice(0, 5).map((row) => ({
+          updateOne: {
+            filter: { _id: rowKey(key, row) },
+            update: { $set: { lastWakeAt: now, lastWakeEventId: wake?._id ?? null } },
+            upsert: true,
+          },
+        })));
+      } catch (stateError) {
+        // The wake is queued. Without its row record a self-set row can wake once more.
+        console.error('[assigned-task-wake] row state not recorded:', key, (stateError as Error).message);
+      }
     } catch (error) {
       if ((error as { code?: number }).code === 11000 && !reserved) continue;
       if (reserved && !enqueueStarted) {
diff --git a/backend/services/nativeRuntimeService.ts b/backend/services/nativeRuntimeService.ts
index ca48af7ca7..7147175d56 100644
--- a/backend/services/nativeRuntimeService.ts
+++ b/backend/services/nativeRuntimeService.ts
@@ -80,6 +80,8 @@ export interface NativeRunResult {
   finalMessage?: string;
   errorKind?: string;
   errorMessage?: string;
+  /** Why a succeeded run did nothing, when the kernel stopped it. */
+  noActionReason?: string;
 }
 
 // --- tuning constants ------------------------------------------------------
@@ -646,6 +648,11 @@ interface DispatchContext {
   webSearches?: number;
   /** Page reads this turn has sent (TASK-267b). */
   webFetches?: number;
+  /**
+   * Set when a post was refused because this agent already holds the floor.
+   * Nothing else it can do this turn changes that, so the run ends after the round.
+   */
+  floorHeld?: boolean;
   /**
    * The addresses written in text the turn was given, never by the model: the
    * prompt, the message it answers, other authors' pod messages, search
@@ -811,6 +818,21 @@ export async function dispatchTool(
           runId: ctx.runId,
           ...(ctx.runId ? { runPodId: String(ctx.podId) } : {}),
         });
+        if (result?.refused && result.reason === 'consecutive_run_cap') {
+          // The shared guidance offers an attachment, which a web-tainted turn
+          // cannot make, and only another speaker clears the cap. Say only that.
+          ctx.floorHeld = true;
+          return {
+            content: {
+              ok: false,
+              posted: false,
+              reason: result.reason,
+              guidance: `Not posted: you have already sent ${result.consecutive} messages in a row here `
+                + 'with nobody else speaking. Wait for someone else to speak. This turn ends now.',
+            },
+            error: 'message_post_refused',
+          };
+        }
         if (result?.refused) {
           return { content: { ok: false, posted: false, result }, error: 'message_post_refused' };
         }
@@ -2274,6 +2296,12 @@ export async function runAgent(
           }
           break;
         }
+        // Only another speaker clears a held floor, so another round would
+        // only spend tokens: end the turn as a decline.
+        if (dispatchCtx.floorHeld) {
+          run.status = run.status === 'running' ? 'succeeded' : run.status;
+          break;
+        }
         // A hosted agent's greeting is one post. Given another round, the
         // model posted the greeting's last line as a second message (QA walk
         // 4, 2026-10-16), so its first-contact run ends at its first post.
@@ -2416,6 +2444,7 @@ export async function runAgent(
     finalMessage,
     errorKind: run.errorKind,
     errorMessage: run.errorMessage,
+    ...(dispatchCtx.floorHeld && !postedViaTool ? { noActionReason: 'consecutive_run_cap' } : {}),
   };
 }
 

From ba56f59cc8255e90e16ba16b4933982f13a5c9df Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:16:59 -0700
Subject: [PATCH 2/7] fix(tasks): legacy agent accounts don't re-arm a self-set
 row; the stop rule fails closed

- A person is now a user with no isBot, no botType and no botMetadata.agentName.
  The two markers come from carriesAgentMarkers, which is lifted out of
  isAgentOwnedRow.
- Pod activity is read as every author after the cutoff (authorIdsSince on PG,
  with a Mongo fallback), not the last 50 messages, and once per cutoff per
  dispatch.
- Rows record lastWakeAt before the wake is queued. If that write fails,
  nothing is queued.
- The stop note is retried. If it still fails, the stop is undone, so the next
  pass writes the note, and the row doesn't wake in between.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.test.js      | 63 +++++++++++++++++++
 .../models/pg/message.activityHint.test.js    | 19 ++++++
 backend/models/pg/Message.ts                  | 12 ++++
 backend/services/agentIdentityService.ts      |  8 ++-
 backend/services/agentMessageService.ts       | 11 ++++
 backend/services/assignedTaskWakeService.ts   | 59 +++++++++++------
 6 files changed, 151 insertions(+), 21 deletions(-)

diff --git a/backend/__tests__/service/assignedTaskWakeRows.test.js b/backend/__tests__/service/assignedTaskWakeRows.test.js
index 11bc5edd60..2b592f729d 100644
--- a/backend/__tests__/service/assignedTaskWakeRows.test.js
+++ b/backend/__tests__/service/assignedTaskWakeRows.test.js
@@ -6,6 +6,7 @@ const Task = require('../../models/Task');
 const User = require('../../models/User');
 const Message = require('../../models/Message');
 const AgentEvent = require('../../models/AgentEvent');
+const AgentTaskRowWakeState = require('../../models/AgentTaskRowWakeState');
 const { AgentInstallation } = require('../../models/AgentRegistry');
 const AgentEventService = require('../../services/agentEventService');
 const { dispatchAssignedTaskWakes } = require('../../services/assignedTaskWakeService');
@@ -94,6 +95,44 @@ describe('a row the agent created for itself', () => {
     expect(await wakes(62)).toBe(0);
   });
 
+  // Agent accounts from before `isBot` carry only the install markers.
+  it.each([
+    ['botType', { botType: 'agent' }],
+    ['botMetadata.agentName', { botMetadata: { agentName: 'openclaw', instanceId: 'aria' } }],
+  ])('is not re-armed by a legacy agent account marked only by %s', async (_marker, markers) => {
+    const { insertedId } = await User.collection.insertOne({ username: 'legacy-aria', ...markers });
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    await say(insertedId, 10);
+    expect(await wakes(31)).toBe(0);
+  });
+
+  it('sees a person who posted before a long run of other agents', async () => {
+    const other = await User.create({
+      username: 'other-bot', email: 'other-bot@example.test', password: 'test-password', isBot: true, botType: 'agent',
+    });
+    expect(await wakes(0)).toBe(1);
+    await settle('succeeded', 1);
+    await say(person._id, 2);
+    await Message.collection.insertMany(Array.from({ length: 60 }, (_, i) => ({
+      podId, userId: other._id, content: `status ${i}`, messageType: 'text', createdAt: new Date(at(3).getTime() + i * 1000),
+    })));
+    expect(await wakes(31)).toBe(1);
+  });
+
+  it('queues no wake it could not record, and wakes once the record works', async () => {
+    const write = jest.spyOn(AgentTaskRowWakeState, 'bulkWrite').mockRejectedValueOnce(new Error('mongo down'));
+    try {
+      expect(await wakes(0)).toBe(0);
+      expect(AgentEventService.enqueue).not.toHaveBeenCalled();
+      expect(await wakes(1)).toBe(1);
+      await settle('succeeded', 2);
+      expect(await wakes(32)).toBe(0);
+    } finally {
+      write.mockRestore();
+    }
+  });
+
   it('is re-armed by a person creating or updating a row in the pod', async () => {
     expect(await wakes(0)).toBe(1);
     await settle('succeeded', 1);
@@ -141,6 +180,30 @@ describe('a row a person created or assigned', () => {
     expect(stopped.updatedAt).toEqual(before);
   });
 
+  it('stays quiet when the stop note cannot be written, and the next pass writes it once', async () => {
+    const realUpdate = Task.updateOne.bind(Task);
+    let refusals = 2;
+    const spy = jest.spyOn(Task, 'updateOne').mockImplementation((filter, update, ...rest) => {
+      if (update?.$push?.updates?.text === STOPPED && refusals > 0) {
+        refusals -= 1;
+        return Promise.reject(new Error('write refused'));
+      }
+      return realUpdate(filter, update, ...rest);
+    });
+    try {
+      for (const minutes of [0, 31, 62]) {
+        await wakes(minutes); // eslint-disable-line no-await-in-loop
+        await settle('failed', minutes + 1); // eslint-disable-line no-await-in-loop
+      }
+      expect(await wakes(93)).toBe(0);
+      expect((await row()).updates || []).toEqual([]);
+      expect(await wakes(124)).toBe(0);
+      expect((await row()).updates.filter((update) => update.text === STOPPED)).toHaveLength(1);
+    } finally {
+      spy.mockRestore();
+    }
+  });
+
   it('starts again after a person updates the stopped row', async () => {
     for (const minutes of [0, 31, 62]) {
       await wakes(minutes); // eslint-disable-line no-await-in-loop
diff --git a/backend/__tests__/unit/models/pg/message.activityHint.test.js b/backend/__tests__/unit/models/pg/message.activityHint.test.js
index ac8694540a..76f683d7b8 100644
--- a/backend/__tests__/unit/models/pg/message.activityHint.test.js
+++ b/backend/__tests__/unit/models/pg/message.activityHint.test.js
@@ -85,3 +85,22 @@ describe('Message.findActivityHint', () => {
     expect(quiet).not.toEqual(broken);
   });
 });
+
+describe('Message.findAuthorIdsSince', () => {
+  beforeEach(() => pool.query.mockReset());
+
+  it('reads every author after the cutoff, with no row limit', async () => {
+    pool.query.mockResolvedValue({ rows: [{ user_id: 'a' }, { user_id: 'b' }] });
+    expect(await Message.findAuthorIdsSince(POD, SINCE)).toEqual(['a', 'b']);
+    const [sql, params] = pool.query.mock.calls[0];
+    expect(sql).toMatch(/DISTINCT user_id/);
+    expect(sql).toMatch(/created_at > \$2/);
+    expect(sql).not.toMatch(/LIMIT/);
+    expect(params).toEqual([POD, SINCE]);
+  });
+
+  it('throws when the store cannot be read, so the caller can tell nobody from unknown', async () => {
+    pool.query.mockRejectedValue(new Error('pg down'));
+    await expect(Message.findAuthorIdsSince(POD, SINCE)).rejects.toThrow('pg down');
+  });
+});
diff --git a/backend/models/pg/Message.ts b/backend/models/pg/Message.ts
index 05f7c9b6f2..907ba940c0 100644
--- a/backend/models/pg/Message.ts
+++ b/backend/models/pg/Message.ts
@@ -676,6 +676,18 @@ class Message {
     }
   }
 
+  // Everyone who posted in the pod after `since`, however many messages that
+  // was. Throws, so a caller can tell "nobody" from "could not read".
+  static async findAuthorIdsSince(podId: unknown, since: Date): Promise<string[]> {
+    const result = await (pool as PgPool).query(
+      `SELECT DISTINCT user_id
+         FROM messages
+        WHERE pod_id = $1 AND created_at > $2 AND message_type != 'system'`,
+      [String(podId), since],
+    );
+    return (result.rows as Array<{ user_id: string }>).map((r) => String(r.user_id));
+  }
+
   static async findMostRecentPodActivity(
     podIds: unknown[],
     since: unknown,
diff --git a/backend/services/agentIdentityService.ts b/backend/services/agentIdentityService.ts
index 39c0fffb77..2f055cf181 100644
--- a/backend/services/agentIdentityService.ts
+++ b/backend/services/agentIdentityService.ts
@@ -161,6 +161,11 @@ export class AgentDisplayNameTakenError extends Error {
  * refusal below could never fire. `agentName` is the leaf this writer always
  * sets (`getOrCreateAgentUser`, create and upgrade arms alike).
  */
+export const carriesAgentMarkers = (row: {
+  botType?: string | null;
+  botMetadata?: { agentName?: string | null } | null;
+}): boolean => Boolean(row.botType) || Boolean(row.botMetadata?.agentName);
+
 export const isAgentOwnedRow = (
   row: {
     email?: string | null;
@@ -169,8 +174,7 @@ export const isAgentOwnedRow = (
   },
   agentEmail: string,
 ): boolean => String(row.email || '').toLowerCase() === agentEmail.toLowerCase()
-  || Boolean(row.botType)
-  || Boolean(row.botMetadata?.agentName);
+  || carriesAgentMarkers(row);
 
 interface AgentTypeConfig {
   officialDisplayName: string;
diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts
index 1019f04858..b3ddbee7c5 100644
--- a/backend/services/agentMessageService.ts
+++ b/backend/services/agentMessageService.ts
@@ -2211,6 +2211,17 @@ class AgentMessageService {
    * exists without changing this array-shaped service contract for existing
    * internal consumers.
    */
+  static async authorIdsSince(podId: unknown, since: Date): Promise<string[]> {
+    if (PGMessage && process.env.PG_HOST) {
+      return (PGMessage as { findAuthorIdsSince(id: unknown, at: Date): Promise<string[]> })
+        .findAuthorIdsSince(podId, since);
+    }
+    const ids: unknown[] = await Message.distinct('userId', {
+      podId: { $eq: podId }, createdAt: { $gt: since }, messageType: { $ne: 'system' },
+    });
+    return ids.map(String);
+  }
+
   static async getRecentMessages(
     podId: unknown,
     limit = 20,
diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts
index c95d9a32cf..5886aa2d4f 100644
--- a/backend/services/assignedTaskWakeService.ts
+++ b/backend/services/assignedTaskWakeService.ts
@@ -7,6 +7,7 @@ import Task from '../models/Task';
 import User from '../models/User';
 import Pod from '../models/Pod';
 import { AgentInstallation } from '../models/AgentRegistry';
+import { carriesAgentMarkers } from './agentIdentityService';
 
 const DELAY_MS = 5 * 60 * 1000;
 const INTERVAL_MS = 30 * 60 * 1000;
@@ -46,10 +47,16 @@ function selfCreated(row: Row, install: Installation, principalId: string): bool
   return Boolean(minted) && normalized(minted?.[1]) === normalized(install.agentName);
 }
 
+// Agent rows written before `isBot` existed carry only the install markers,
+// so a person is a row with neither the flag nor those markers.
 async function anyHuman(ids: unknown[]): Promise<boolean> {
   const unique = [...new Set(ids.map((id) => String(id || '')))].filter((id) => Types.ObjectId.isValid(id));
   if (!unique.length) return false;
-  return Boolean(await User.exists({ _id: { $in: unique }, isBot: { $ne: true } }));
+  const users = await User.find({ _id: { $in: unique }, isBot: { $ne: true } })
+    .select('botType botMetadata.agentName').lean() as Array<{
+      botType?: string | null; botMetadata?: { agentName?: string | null } | null;
+    }>;
+  return users.some((user) => !carriesAgentMarkers(user));
 }
 
 const updatedSince = (row: Row, since: Date): unknown[] => (row.updates || [])
@@ -62,11 +69,7 @@ async function humanActedInPodSince(podId: unknown, since: Date): Promise<boolea
   const AgentMessageService = require('./agentMessageService');
   let authors: unknown[] = [];
   try {
-    const recent = await AgentMessageService.getRecentMessages(podId, 50) as Array<{
-      createdAt?: unknown; userId?: { _id?: unknown } | unknown; user_id?: unknown;
-    }>;
-    authors = recent.filter((message) => new Date(String(message.createdAt || 0)) > since)
-      .map((message) => (message.userId as { _id?: unknown })?._id || message.userId || message.user_id);
+    authors = await AgentMessageService.authorIdsSince(podId, since);
   } catch (error) {
     // Unread means no wake: a row the agent set itself waits for a person anyway.
     console.warn('[assigned-task-wake] pod activity unreadable:', String(podId), (error as Error).message);
@@ -120,12 +123,20 @@ async function stopRow(row: Row, state: RowState, install: Installation, princip
     { $set: { stoppedAt: now } },
   );
   if (!stopped.modifiedCount) return;
-  state.stoppedAt = now;
-  await Task.updateOne(
+  const writeNote = () => Task.updateOne(
     { _id: row._id },
     { $push: { updates: { text: STOPPED_NOTE, author: install.agentName, authorId: principalId, createdAt: now } } },
     { timestamps: false },
   );
+  try {
+    await writeNote().catch(writeNote);
+  } catch (error) {
+    // Un-stop so the next dispatch says so again, rather than a silent stop.
+    await AgentTaskRowWakeState.updateOne({ _id: state._id, stoppedAt: now }, { $set: { stoppedAt: null } });
+    console.error('[assigned-task-wake] stop note not written:', state._id, (error as Error).message);
+    return;
+  }
+  state.stoppedAt = now;
 }
 
 // The rows that may wake this seat now. A row the agent created for itself
@@ -139,12 +150,18 @@ async function wakeableRows(
     .lean() as RowState[];
   await countOutcomes(states);
   const byRow = new Map(states.map((state) => [state._id, state]));
+  // Every row here is in the install's pod; read its activity once per cutoff.
+  const actedSince = new Map<number, Promise<boolean>>();
+  const humanActedSince = (since: Date): Promise<boolean> => {
+    if (!actedSince.has(since.getTime())) actedSince.set(since.getTime(), humanActedInPodSince(install.podId, since));
+    return actedSince.get(since.getTime()) as Promise<boolean>;
+  };
   const wakeable: Row[] = [];
   for (const row of owned) {
     const state = byRow.get(rowKey(seat, row));
     if (selfCreated(row, install, principalId)) {
       // eslint-disable-next-line no-await-in-loop
-      if (!state?.lastWakeAt || await humanActedInPodSince(row.podId, new Date(state.lastWakeAt))) wakeable.push(row);
+      if (!state?.lastWakeAt || await humanActedSince(new Date(state.lastWakeAt))) wakeable.push(row);
       continue;
     }
     if (!state) {
@@ -302,6 +319,14 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
         await AgentTaskWakeState.updateOne({ _id: key, lastWakeAt: now }, { $unset: { lastWakeAt: 1 } });
         continue;
       }
+      const shown = current.slice(0, 5);
+      const recordRows = (update: Record<string, unknown>) => AgentTaskRowWakeState.bulkWrite(shown.map((row) => ({
+        updateOne: { filter: { _id: rowKey(key, row) }, update: { $set: update }, upsert: true },
+      })));
+      // Recorded before the wake is queued: a wake the stop rule cannot see
+      // must not happen. On failure the slot is returned below and nothing queues.
+      // eslint-disable-next-line no-await-in-loop
+      await recordRows({ lastWakeAt: now, lastWakeEventId: null });
       // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
       const AgentEventService = require('./agentEventService');
       enqueueStarted = true;
@@ -312,25 +337,21 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
         payload: {
           triggerAuthor: 'kernel', assignedTaskWake: true,
           content: '[Your assigned work is still open]\n\n'
-            + current.slice(0, 5).map((row) => `- ${row.taskId}: ${row.title.slice(0, 220)}`).join('\n')
+            + shown.map((row) => `- ${row.taskId}: ${row.title.slice(0, 220)}`).join('\n')
             + '\n\nContinue the next actionable row. Recheck its current state and claim it before starting. '
             + 'If you are waiting on a dependency or operator gate, mark the row blocked with the reason. '
             + 'Post only an action or a materially useful result.',
         },
       });
       woken += 1;
+      const linkWake = () => recordRows({ lastWakeEventId: wake?._id ?? null });
       try {
         // eslint-disable-next-line no-await-in-loop
-        await AgentTaskRowWakeState.bulkWrite(current.slice(0, 5).map((row) => ({
-          updateOne: {
-            filter: { _id: rowKey(key, row) },
-            update: { $set: { lastWakeAt: now, lastWakeEventId: wake?._id ?? null } },
-            upsert: true,
-          },
-        })));
+        await linkWake().catch(linkWake);
       } catch (stateError) {
-        // The wake is queued. Without its row record a self-set row can wake once more.
-        console.error('[assigned-task-wake] row state not recorded:', key, (stateError as Error).message);
+        // The rows already carry lastWakeAt, so a self-set row still waits for a
+        // person. Only this one run's outcome goes uncounted toward the stop.
+        console.error('[assigned-task-wake] wake not linked to rows:', key, (stateError as Error).message);
       }
     } catch (error) {
       if ((error as { code?: number }).code === 11000 && !reserved) continue;

From 2b5ff8d911ecb2a9c0d3c8cc9998e519177fe695 Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:24:18 -0700
Subject: [PATCH 3/7] fix(tasks): the derived agent address is an agent marker
 too

carriesAgentMarkers now covers all three markers an install writes (the
agents.commonly.local address, botType, botMetadata.agentName), so anyHuman
and isAgentOwnedRow read one predicate. An email-only legacy agent no longer
re-arms a self-set row.

The capped-post test now posts first and is then refused in the same round,
so it exercises the postedViaTool guard.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.test.js        |  1 +
 .../nativeRuntimeService.webSearch.test.js      | 17 +++++++++++++----
 backend/services/agentIdentityService.ts        |  5 ++++-
 backend/services/assignedTaskWakeService.ts     | 11 +++++++----
 4 files changed, 25 insertions(+), 9 deletions(-)

diff --git a/backend/__tests__/service/assignedTaskWakeRows.test.js b/backend/__tests__/service/assignedTaskWakeRows.test.js
index 2b592f729d..0c3b86523a 100644
--- a/backend/__tests__/service/assignedTaskWakeRows.test.js
+++ b/backend/__tests__/service/assignedTaskWakeRows.test.js
@@ -99,6 +99,7 @@ describe('a row the agent created for itself', () => {
   it.each([
     ['botType', { botType: 'agent' }],
     ['botMetadata.agentName', { botMetadata: { agentName: 'openclaw', instanceId: 'aria' } }],
+    ['its derived address', { email: 'openclaw-aria@agents.commonly.local' }],
   ])('is not re-armed by a legacy agent account marked only by %s', async (_marker, markers) => {
     const { insertedId } = await User.collection.insertOne({ username: 'legacy-aria', ...markers });
     expect(await wakes(0)).toBe(1);
diff --git a/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js b/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
index e02a96fe13..db6465433f 100644
--- a/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
+++ b/backend/__tests__/unit/services/nativeRuntimeService.webSearch.test.js
@@ -164,12 +164,21 @@ describe('native turn with commonly_web_search', () => {
     AgentMessageService.postMessage.mockReset().mockResolvedValue({ success: true });
   });
 
-  test('a run that posts after an earlier capped refusal reports no decline', async () => {
-    AgentMessageService.postMessage.mockResolvedValueOnce({ success: true, message: { id: 'posted-1' } });
-    axios.post.mockResolvedValueOnce(toolCalls(call('p1', 'commonly_post_message', { content: 'Done.' })))
-      .mockResolvedValueOnce(finalReply);
+  test('a run that posted before its capped refusal is not reported as a decline', async () => {
+    const capped = { success: false, refused: true, reason: 'consecutive_run_cap', consecutive: 3 };
+    AgentMessageService.postMessage.mockResolvedValueOnce({ success: true, message: { id: 'posted-1' } })
+      .mockResolvedValueOnce(capped);
+    axios.post.mockResolvedValueOnce(toolCalls(
+      call('p1', 'commonly_post_message', { content: 'Done.' }),
+      call('p2', 'commonly_post_message', { content: 'One more thing.' }),
+    )).mockResolvedValueOnce(finalReply);
     const result = await runAgent(install, mention);
+    expect(AgentMessageService.postMessage).toHaveBeenCalledTimes(2);
+    expect(savedRun.turns[0].toolCalls[1]).toEqual(expect.objectContaining({ error: 'message_post_refused' }));
+    expect(result.status).toBe('succeeded');
     expect(result.noActionReason).toBeUndefined();
+    axios.post.mockReset();
+    AgentMessageService.postMessage.mockReset().mockResolvedValue({ success: true });
   });
 
   test('results reach the model marked untrusted; the run keeps the count and the addresses only', async () => {
diff --git a/backend/services/agentIdentityService.ts b/backend/services/agentIdentityService.ts
index 2f055cf181..a4b5b98ff3 100644
--- a/backend/services/agentIdentityService.ts
+++ b/backend/services/agentIdentityService.ts
@@ -162,9 +162,12 @@ export class AgentDisplayNameTakenError extends Error {
  * sets (`getOrCreateAgentUser`, create and upgrade arms alike).
  */
 export const carriesAgentMarkers = (row: {
+  email?: string | null;
   botType?: string | null;
   botMetadata?: { agentName?: string | null } | null;
-}): boolean => Boolean(row.botType) || Boolean(row.botMetadata?.agentName);
+}): boolean => isReservedAgentEmail(row.email)
+  || Boolean(row.botType)
+  || Boolean(row.botMetadata?.agentName);
 
 export const isAgentOwnedRow = (
   row: {
diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts
index 5886aa2d4f..0138d86dbb 100644
--- a/backend/services/assignedTaskWakeService.ts
+++ b/backend/services/assignedTaskWakeService.ts
@@ -47,14 +47,15 @@ function selfCreated(row: Row, install: Installation, principalId: string): bool
   return Boolean(minted) && normalized(minted?.[1]) === normalized(install.agentName);
 }
 
-// Agent rows written before `isBot` existed carry only the install markers,
-// so a person is a row with neither the flag nor those markers.
+// Agent rows written before `isBot` existed carry only the install markers
+// (the derived address, botType, agentName), so a person is a row with neither
+// the flag nor any of those markers.
 async function anyHuman(ids: unknown[]): Promise<boolean> {
   const unique = [...new Set(ids.map((id) => String(id || '')))].filter((id) => Types.ObjectId.isValid(id));
   if (!unique.length) return false;
   const users = await User.find({ _id: { $in: unique }, isBot: { $ne: true } })
-    .select('botType botMetadata.agentName').lean() as Array<{
-      botType?: string | null; botMetadata?: { agentName?: string | null } | null;
+    .select('email botType botMetadata.agentName').lean() as Array<{
+      email?: string | null; botType?: string | null; botMetadata?: { agentName?: string | null } | null;
     }>;
   return users.some((user) => !carriesAgentMarkers(user));
 }
@@ -325,6 +326,8 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
       })));
       // Recorded before the wake is queued: a wake the stop rule cannot see
       // must not happen. On failure the slot is returned below and nothing queues.
+      // A partial failure can leave some rows with lastWakeAt and no event id:
+      // they wait for a person as usual and this run is not counted for them.
       // eslint-disable-next-line no-await-in-loop
       await recordRows({ lastWakeAt: now, lastWakeEventId: null });
       // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports

From 390a6f357bf61b1b10ce133101515e2519a5806b Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:27:54 -0700
Subject: [PATCH 4/7] fix(tasks): keep the agent-address check out of the
 adoption guard

isAgentOwnedRow guards account adoption, so there only this seat's own
derived address counts. It goes back to how it was at ba56f59c. anyHuman
checks the agents.commonly.local domain itself; erring toward "agent" there
only keeps a self-set row quiet.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 backend/services/agentIdentityService.ts    | 5 +----
 backend/services/assignedTaskWakeService.ts | 8 +++++---
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/backend/services/agentIdentityService.ts b/backend/services/agentIdentityService.ts
index a4b5b98ff3..2f055cf181 100644
--- a/backend/services/agentIdentityService.ts
+++ b/backend/services/agentIdentityService.ts
@@ -162,12 +162,9 @@ export class AgentDisplayNameTakenError extends Error {
  * sets (`getOrCreateAgentUser`, create and upgrade arms alike).
  */
 export const carriesAgentMarkers = (row: {
-  email?: string | null;
   botType?: string | null;
   botMetadata?: { agentName?: string | null } | null;
-}): boolean => isReservedAgentEmail(row.email)
-  || Boolean(row.botType)
-  || Boolean(row.botMetadata?.agentName);
+}): boolean => Boolean(row.botType) || Boolean(row.botMetadata?.agentName);
 
 export const isAgentOwnedRow = (
   row: {
diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts
index 0138d86dbb..d921f20c89 100644
--- a/backend/services/assignedTaskWakeService.ts
+++ b/backend/services/assignedTaskWakeService.ts
@@ -7,7 +7,7 @@ import Task from '../models/Task';
 import User from '../models/User';
 import Pod from '../models/Pod';
 import { AgentInstallation } from '../models/AgentRegistry';
-import { carriesAgentMarkers } from './agentIdentityService';
+import { carriesAgentMarkers, isReservedAgentEmail } from './agentIdentityService';
 
 const DELAY_MS = 5 * 60 * 1000;
 const INTERVAL_MS = 30 * 60 * 1000;
@@ -49,7 +49,9 @@ function selfCreated(row: Row, install: Installation, principalId: string): bool
 
 // Agent rows written before `isBot` existed carry only the install markers
 // (the derived address, botType, agentName), so a person is a row with neither
-// the flag nor any of those markers.
+// the flag nor any of those markers. The address is checked as a domain here,
+// not through isAgentOwnedRow: that guards adoption, where only this seat's own
+// address counts. Erring toward "agent" only keeps a self-set row quiet.
 async function anyHuman(ids: unknown[]): Promise<boolean> {
   const unique = [...new Set(ids.map((id) => String(id || '')))].filter((id) => Types.ObjectId.isValid(id));
   if (!unique.length) return false;
@@ -57,7 +59,7 @@ async function anyHuman(ids: unknown[]): Promise<boolean> {
     .select('email botType botMetadata.agentName').lean() as Array<{
       email?: string | null; botType?: string | null; botMetadata?: { agentName?: string | null } | null;
     }>;
-  return users.some((user) => !carriesAgentMarkers(user));
+  return users.some((user) => !isReservedAgentEmail(user.email) && !carriesAgentMarkers(user));
 }
 
 const updatedSince = (row: Row, since: Date): unknown[] => (row.updates || [])

From 49ff10d143af6a48a97c41ed3452584594f4bf1a Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:41:25 -0700
Subject: [PATCH 5/7] fix(tasks): read row updates before the pod read, read
 both stores, clear a partial stamp

- humanActedInPodSince checks row updates first, so a failed message read no
  longer hides a person's row update. A failed message read still fails
  closed: re-arming on an unreadable pod would bring back the loop.
- authorIdsSince on PostgreSQL also reads MongoDB, where a fallback post sits
  until it is reconciled.
- When the pre-enqueue row write fails part way, the rows it stamped get
  lastWakeAt back to null, so their first wake is not lost.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.test.js      | 49 +++++++++++++++++++
 ...agentMessageService.authorIdsSince.test.js | 48 ++++++++++++++++++
 backend/services/agentMessageService.ts       | 17 ++++---
 backend/services/assignedTaskWakeService.ts   | 30 ++++++++----
 4 files changed, 127 insertions(+), 17 deletions(-)
 create mode 100644 backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js

diff --git a/backend/__tests__/service/assignedTaskWakeRows.test.js b/backend/__tests__/service/assignedTaskWakeRows.test.js
index 0c3b86523a..15108b8de9 100644
--- a/backend/__tests__/service/assignedTaskWakeRows.test.js
+++ b/backend/__tests__/service/assignedTaskWakeRows.test.js
@@ -9,6 +9,7 @@ const AgentEvent = require('../../models/AgentEvent');
 const AgentTaskRowWakeState = require('../../models/AgentTaskRowWakeState');
 const { AgentInstallation } = require('../../models/AgentRegistry');
 const AgentEventService = require('../../services/agentEventService');
+const AgentMessageService = require('../../services/agentMessageService');
 const { dispatchAssignedTaskWakes } = require('../../services/assignedTaskWakeService');
 
 const STOPPED = 'I stopped retrying this after repeated errors. Reassign it or update it to try again.';
@@ -134,6 +135,54 @@ describe('a row the agent created for itself', () => {
     }
   });
 
+  it('stays quiet when the pod read fails and nobody updated a row', async () => {
+    const read = jest.spyOn(AgentMessageService, 'authorIdsSince').mockRejectedValue(new Error('pg down'));
+    try {
+      expect(await wakes(0)).toBe(1);
+      await settle('succeeded', 1);
+      await say(person._id, 10);
+      expect(await wakes(31)).toBe(0);
+    } finally {
+      read.mockRestore();
+    }
+  });
+
+  it('is still re-armed by a person updating a row when the pod read fails', async () => {
+    const read = jest.spyOn(AgentMessageService, 'authorIdsSince').mockRejectedValue(new Error('pg down'));
+    try {
+      expect(await wakes(0)).toBe(1);
+      await settle('succeeded', 1);
+      await Task.create({
+        podId, taskNum: 5, taskId: 'TASK-5', title: 'Unrelated', status: 'done',
+        updates: [{ text: 'Created by row-owner', author: 'row-owner', authorId: String(person._id), createdAt: at(10) }],
+      });
+      expect(await wakes(31)).toBe(1);
+    } finally {
+      read.mockRestore();
+    }
+  });
+
+  it('clears the stamp on rows a failed write reached, so their first wake is not lost', async () => {
+    await Task.create({
+      podId, taskNum: 6, taskId: 'N-6-codex-mvid77ql', title: 'Second own row', assignee: 'forge', source: 'agent',
+    });
+    const original = AgentTaskRowWakeState.bulkWrite.bind(AgentTaskRowWakeState);
+    const write = jest.spyOn(AgentTaskRowWakeState, 'bulkWrite').mockImplementationOnce(async (ops) => {
+      await original(ops.slice(0, 1));
+      throw new Error('write failed after the first row');
+    });
+    try {
+      expect(await wakes(0)).toBe(0);
+      expect(AgentEventService.enqueue).not.toHaveBeenCalled();
+      expect(await wakes(1)).toBe(1);
+      const { content } = AgentEventService.enqueue.mock.calls[0][0].payload;
+      expect(content).toContain('N-4-codex-mvid77qk');
+      expect(content).toContain('N-6-codex-mvid77ql');
+    } finally {
+      write.mockRestore();
+    }
+  });
+
   it('is re-armed by a person creating or updating a row in the pod', async () => {
     expect(await wakes(0)).toBe(1);
     await settle('succeeded', 1);
diff --git a/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js b/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js
new file mode 100644
index 0000000000..37a6a1650f
--- /dev/null
+++ b/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js
@@ -0,0 +1,48 @@
+/**
+ * Who posted in a pod since a time: the assigned-task wake reads this to decide
+ * whether a person acted. On PostgreSQL a post that fell back to MongoDB is only
+ * in MongoDB until it is reconciled, so both stores are read.
+ */
+const AgentMessageService = require('../../../services/agentMessageService');
+const Message = require('../../../models/Message');
+const PGMessage = require('../../../models/pg/Message');
+
+jest.mock('../../../models/Message');
+jest.mock('../../../models/pg/Message', () => ({ findAuthorIdsSince: jest.fn() }));
+
+const SINCE = new Date('2026-10-24T11:00:00Z');
+
+describe('authorIdsSince', () => {
+  const origPgHost = process.env.PG_HOST;
+  beforeEach(() => {
+    Message.distinct.mockReset();
+    PGMessage.findAuthorIdsSince.mockReset();
+  });
+  afterAll(() => {
+    if (origPgHost === undefined) delete process.env.PG_HOST;
+    else process.env.PG_HOST = origPgHost;
+  });
+
+  test('on PostgreSQL, includes an author whose post is still in the MongoDB fallback', async () => {
+    process.env.PG_HOST = 'localhost';
+    PGMessage.findAuthorIdsSince.mockResolvedValue(['agent-1', 'person-1']);
+    Message.distinct.mockResolvedValue(['person-2', 'person-1']);
+    const ids = await AgentMessageService.authorIdsSince('pod-1', SINCE);
+    expect(ids.sort()).toEqual(['agent-1', 'person-1', 'person-2']);
+    expect(Message.distinct).toHaveBeenCalledWith('userId', expect.objectContaining({ createdAt: { $gt: SINCE } }));
+  });
+
+  test('a failed PostgreSQL read is not hidden by the fallback read', async () => {
+    process.env.PG_HOST = 'localhost';
+    PGMessage.findAuthorIdsSince.mockRejectedValue(new Error('pg down'));
+    Message.distinct.mockResolvedValue(['person-2']);
+    await expect(AgentMessageService.authorIdsSince('pod-1', SINCE)).rejects.toThrow('pg down');
+  });
+
+  test('without PostgreSQL, reads MongoDB only', async () => {
+    delete process.env.PG_HOST;
+    Message.distinct.mockResolvedValue(['person-2']);
+    expect(await AgentMessageService.authorIdsSince('pod-1', SINCE)).toEqual(['person-2']);
+    expect(PGMessage.findAuthorIdsSince).not.toHaveBeenCalled();
+  });
+});
diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts
index b3ddbee7c5..2e300330bf 100644
--- a/backend/services/agentMessageService.ts
+++ b/backend/services/agentMessageService.ts
@@ -2212,14 +2212,17 @@ class AgentMessageService {
    * internal consumers.
    */
   static async authorIdsSince(podId: unknown, since: Date): Promise<string[]> {
-    if (PGMessage && process.env.PG_HOST) {
-      return (PGMessage as { findAuthorIdsSince(id: unknown, at: Date): Promise<string[]> })
-        .findAuthorIdsSince(podId, since);
-    }
-    const ids: unknown[] = await Message.distinct('userId', {
+    const mongo = Message.distinct('userId', {
       podId: { $eq: podId }, createdAt: { $gt: since }, messageType: { $ne: 'system' },
-    });
-    return ids.map(String);
+    }) as Promise<unknown[]>;
+    if (!PGMessage || !process.env.PG_HOST) return (await mongo).map(String);
+    // A post that fell back to MongoDB while PostgreSQL was failing is only
+    // there until it is reconciled, so read both.
+    const [pg, fallback] = await Promise.all([
+      (PGMessage as { findAuthorIdsSince(id: unknown, at: Date): Promise<string[]> }).findAuthorIdsSince(podId, since),
+      mongo,
+    ]);
+    return [...new Set([...pg, ...fallback.map(String)])];
   }
 
   static async getRecentMessages(
diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts
index d921f20c89..e9826ba9b8 100644
--- a/backend/services/assignedTaskWakeService.ts
+++ b/backend/services/assignedTaskWakeService.ts
@@ -70,17 +70,18 @@ const updatedSince = (row: Row, since: Date): unknown[] => (row.updates || [])
 async function humanActedInPodSince(podId: unknown, since: Date): Promise<boolean> {
   // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
   const AgentMessageService = require('./agentMessageService');
-  let authors: unknown[] = [];
+  const rows = await Task.find({ podId, 'updates.createdAt': { $gt: since } })
+    .select('updates.authorId updates.createdAt').lean() as Row[];
+  if (await anyHuman(rows.flatMap((row) => updatedSince(row, since)))) return true;
   try {
-    authors = await AgentMessageService.authorIdsSince(podId, since);
+    return await anyHuman(await AgentMessageService.authorIdsSince(podId, since));
   } catch (error) {
-    // Unread means no wake: a row the agent set itself waits for a person anyway.
+    // Fails closed: re-arming on an unreadable pod would bring back the
+    // thirty-minute loop for as long as the read keeps failing. A missed re-arm
+    // costs one delayed wake; the next update to a row re-arms it.
     console.warn('[assigned-task-wake] pod activity unreadable:', String(podId), (error as Error).message);
     return false;
   }
-  const rows = await Task.find({ podId, 'updates.createdAt': { $gt: since } })
-    .select('updates.authorId updates.createdAt').lean() as Row[];
-  return anyHuman([...authors, ...rows.flatMap((row) => updatedSince(row, since))]);
 }
 
 // 'open' until the wake's run settles; a run that did nothing for its own
@@ -328,10 +329,19 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli
       })));
       // Recorded before the wake is queued: a wake the stop rule cannot see
       // must not happen. On failure the slot is returned below and nothing queues.
-      // A partial failure can leave some rows with lastWakeAt and no event id:
-      // they wait for a person as usual and this run is not counted for them.
-      // eslint-disable-next-line no-await-in-loop
-      await recordRows({ lastWakeAt: now, lastWakeEventId: null });
+      try {
+        // eslint-disable-next-line no-await-in-loop
+        await recordRows({ lastWakeAt: now, lastWakeEventId: null });
+      } catch (recordError) {
+        // An ordered write can stamp some rows before it fails. Those rows were
+        // not woken, so clear the stamp: at worst one extra wake, never a lost one.
+        // eslint-disable-next-line no-await-in-loop
+        await AgentTaskRowWakeState.updateMany(
+          { _id: { $in: shown.map((row) => rowKey(key, row)) }, lastWakeAt: now },
+          { $set: { lastWakeAt: null } },
+        ).catch((clearError: Error) => console.error('[assigned-task-wake] stamp not cleared:', key, clearError.message));
+        throw recordError;
+      }
       // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
       const AgentEventService = require('./agentEventService');
       enqueueStarted = true;

From 26934d98711488ed294c087305f0f29c259676a6 Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 18:02:04 -0700
Subject: [PATCH 6/7] fix(agents): pin the wake-row tests to MongoDB; one
 failed store keeps the other's authors

The Tier 1 job sets PG_HOST, so the floor and author reads went to
PostgreSQL while the fixture wrote MongoDB, and "does not wake an agent
whose last posts in the pod are still unanswered" saw an empty pod.
Reproduced on mongo:7 + postgres:16 with schema.sql loaded; the file now
pins PG_HOST='' like tasks.watchers.test.js.

authorIdsSince (CodeRabbit): a rejected MongoDB read no longer discards
a PostgreSQL result, or the reverse. Both failing still throws, so the
wake fails closed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.test.js      |  5 +++++
 ...agentMessageService.authorIdsSince.test.js | 21 +++++++++++++++++--
 backend/services/agentMessageService.ts       | 14 ++++++++++---
 3 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/backend/__tests__/service/assignedTaskWakeRows.test.js b/backend/__tests__/service/assignedTaskWakeRows.test.js
index 15108b8de9..c2f8918cf3 100644
--- a/backend/__tests__/service/assignedTaskWakeRows.test.js
+++ b/backend/__tests__/service/assignedTaskWakeRows.test.js
@@ -1,3 +1,8 @@
+// The fixture writes pod messages to MongoDB, so the service must read them
+// there. With PG_HOST set (the Tier 1 job), the floor and author reads go to
+// PostgreSQL, which these tests never write; the store read itself is covered
+// in agentMessageService.authorIdsSince.test.js.
+process.env.PG_HOST = '';
 const mongoose = require('mongoose');
 const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils');
 jest.mock('jsonwebtoken', () => ({}));
diff --git a/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js b/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js
index 37a6a1650f..8b6e866bbf 100644
--- a/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js
+++ b/backend/__tests__/unit/services/agentMessageService.authorIdsSince.test.js
@@ -1,7 +1,8 @@
 /**
  * Who posted in a pod since a time: the assigned-task wake reads this to decide
  * whether a person acted. On PostgreSQL a post that fell back to MongoDB is only
- * in MongoDB until it is reconciled, so both stores are read.
+ * in MongoDB until it is reconciled, so both stores are read. One store failing
+ * keeps the other's authors; both failing throws, and the wake fails closed.
  */
 const AgentMessageService = require('../../../services/agentMessageService');
 const Message = require('../../../models/Message');
@@ -32,10 +33,26 @@ describe('authorIdsSince', () => {
     expect(Message.distinct).toHaveBeenCalledWith('userId', expect.objectContaining({ createdAt: { $gt: SINCE } }));
   });
 
-  test('a failed PostgreSQL read is not hidden by the fallback read', async () => {
+  // An author either store returns did post. During a PostgreSQL outage the
+  // posts land in MongoDB, so that read is the one that sees them.
+  test('a failed PostgreSQL read keeps the MongoDB authors', async () => {
     process.env.PG_HOST = 'localhost';
     PGMessage.findAuthorIdsSince.mockRejectedValue(new Error('pg down'));
     Message.distinct.mockResolvedValue(['person-2']);
+    expect(await AgentMessageService.authorIdsSince('pod-1', SINCE)).toEqual(['person-2']);
+  });
+
+  test('a failed MongoDB read keeps the PostgreSQL authors', async () => {
+    process.env.PG_HOST = 'localhost';
+    PGMessage.findAuthorIdsSince.mockResolvedValue(['person-1']);
+    Message.distinct.mockRejectedValue(new Error('mongo down'));
+    expect(await AgentMessageService.authorIdsSince('pod-1', SINCE)).toEqual(['person-1']);
+  });
+
+  test('both reads failing throws', async () => {
+    process.env.PG_HOST = 'localhost';
+    PGMessage.findAuthorIdsSince.mockRejectedValue(new Error('pg down'));
+    Message.distinct.mockRejectedValue(new Error('mongo down'));
     await expect(AgentMessageService.authorIdsSince('pod-1', SINCE)).rejects.toThrow('pg down');
   });
 
diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts
index 2e300330bf..f5618f957e 100644
--- a/backend/services/agentMessageService.ts
+++ b/backend/services/agentMessageService.ts
@@ -2217,12 +2217,20 @@ class AgentMessageService {
     }) as Promise<unknown[]>;
     if (!PGMessage || !process.env.PG_HOST) return (await mongo).map(String);
     // A post that fell back to MongoDB while PostgreSQL was failing is only
-    // there until it is reconciled, so read both.
-    const [pg, fallback] = await Promise.all([
+    // there until it is reconciled, so read both. An author either store
+    // returns did post, so one store failing keeps the other's authors; only
+    // both failing leaves nothing to answer with.
+    const [pg, fallback] = await Promise.allSettled([
       (PGMessage as { findAuthorIdsSince(id: unknown, at: Date): Promise<string[]> }).findAuthorIdsSince(podId, since),
       mongo,
     ]);
-    return [...new Set([...pg, ...fallback.map(String)])];
+    if (pg.status === 'rejected' && fallback.status === 'rejected') throw pg.reason;
+    const ids: string[] = [];
+    for (const [store, read] of [['postgres', pg], ['mongo', fallback]] as const) {
+      if (read.status === 'fulfilled') ids.push(...read.value.map(String));
+      else console.warn(`[agent-message] ${store} author read failed:`, String(podId), (read.reason as Error)?.message);
+    }
+    return [...new Set(ids)];
   }
 
   static async getRecentMessages(

From 8e3fe1a8cf55d2228f9bbf14f55090524757b03f Mon Sep 17 00:00:00 2001
From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com>
Date: Sun, 25 Oct 2026 18:06:16 -0700
Subject: [PATCH 7/7] test(agents): check the wake floor and re-arm against
 PostgreSQL pod messages

Production reads pod messages from PostgreSQL, and the MongoDB-pinned
row tests no longer reach that path. This Tier 1 file writes the
agent's posts and a person's reply through the PG message model and
asserts the floor holds while the posts are unanswered, a person's PG
message re-arms the row, and the agent's own does not. Skipped without
INTEGRATION_TEST and PG_HOST.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013ENddpGzhE7S8qjDSYAMuu
---
 .../service/assignedTaskWakeRows.pg.test.js   | 124 ++++++++++++++++++
 1 file changed, 124 insertions(+)
 create mode 100644 backend/__tests__/service/assignedTaskWakeRows.pg.test.js

diff --git a/backend/__tests__/service/assignedTaskWakeRows.pg.test.js b/backend/__tests__/service/assignedTaskWakeRows.pg.test.js
new file mode 100644
index 0000000000..5442702267
--- /dev/null
+++ b/backend/__tests__/service/assignedTaskWakeRows.pg.test.js
@@ -0,0 +1,124 @@
+// Production reads pod messages from PostgreSQL, so the floor and the re-arm
+// are checked here against real PostgreSQL rows written through the PG model.
+// assignedTaskWakeRows.test.js covers the same rules with MongoDB messages.
+const mongoose = require('mongoose');
+const {
+  setupMongoDb, closeMongoDb, clearMongoDb, setupPgDb, clearPgDb, closePgDb,
+} = require('../utils/testUtils');
+
+jest.mock('jsonwebtoken', () => ({}));
+jest.mock('../../services/agentEventService', () => ({ enqueue: jest.fn() }));
+const { pool } = require('../../config/db-pg');
+const PGMessage = require('../../models/pg/Message');
+const Task = require('../../models/Task');
+const User = require('../../models/User');
+const Message = require('../../models/Message');
+const AgentEvent = require('../../models/AgentEvent');
+const { AgentInstallation } = require('../../models/AgentRegistry');
+const AgentEventService = require('../../services/agentEventService');
+const { dispatchAssignedTaskWakes } = require('../../services/assignedTaskWakeService');
+
+const d = process.env.INTEGRATION_TEST === 'true' && process.env.PG_HOST ? describe : describe.skip;
+let NOW;
+const at = (minutes) => new Date(NOW.getTime() + minutes * 60000);
+let podId;
+let agent;
+let person;
+
+const acked = (when, extra) => ({
+  ...extra,
+  status: 'acked',
+  delivery: { outcome: 'no_action', reasonRecorded: true, updatedAt: when },
+  createdAt: when,
+  updatedAt: when,
+});
+const eventRow = (extra) => ({
+  agentName: 'codex', instanceId: 'forge', podId, type: 'chat.mention', deliveryMode: 'wake', ...extra,
+});
+async function settle(minutes) {
+  const { _id } = await AgentEventService.enqueue.mock.results.at(-1).value;
+  const when = at(minutes);
+  await AgentEvent.collection.insertOne(eventRow(acked(when, { _id })));
+}
+const wakes = async (minutes) => (await dispatchAssignedTaskWakes(at(minutes))).woken;
+const say = (user) => PGMessage.create(String(podId), String(user._id), 'hello');
+
+d('assigned-task wakes read PostgreSQL pod messages', () => {
+  beforeAll(async () => {
+    await setupMongoDb();
+    await setupPgDb();
+  });
+  afterAll(async () => {
+    await clearPgDb();
+    await closePgDb();
+    await pool.end();
+    await closeMongoDb();
+  });
+  beforeEach(async () => {
+    await clearMongoDb();
+    await clearPgDb();
+    podId = new mongoose.Types.ObjectId();
+    agent = await User.create({
+      username: 'forge-pg-row',
+      email: 'forge-pg-row@example.test',
+      password: 'test-password',
+      isBot: true,
+      botType: 'agent',
+      botMetadata: { agentName: 'codex', instanceId: 'forge' },
+    });
+    person = await User.create({
+      username: 'pg-row-owner',
+      email: 'pg-row-owner@example.test',
+      password: 'test-password',
+    });
+    await pool.query(
+      "INSERT INTO pods (id, name, type, created_by) VALUES ($1, 'wake-rows', 'chat', $2)",
+      [String(podId), String(person._id)],
+    );
+    await pool.query(
+      "INSERT INTO users (_id, username, is_bot) VALUES ($1, 'forge-pg-row', true), ($2, 'pg-row-owner', false)",
+      [String(agent._id), String(person._id)],
+    );
+    await AgentInstallation.create({
+      agentName: 'codex',
+      instanceId: 'forge',
+      podId,
+      version: '1',
+      installedBy: agent._id,
+      config: { boardWake: { enabled: false } },
+      status: 'active',
+    });
+    // created_at is the database's clock, so the wakes run on it too.
+    NOW = (await pool.query('SELECT clock_timestamp() AS now')).rows[0].now;
+    await AgentEvent.collection.insertOne(eventRow(acked(at(-6))));
+    AgentEventService.enqueue.mockReset().mockImplementation(async () => ({ _id: new mongoose.Types.ObjectId() }));
+  });
+  afterEach(async () => {
+    // Every message these cases read came from PostgreSQL.
+    expect(await Message.countDocuments({ podId })).toBe(0);
+  });
+
+  it('does not wake an agent whose last posts are unanswered, and wakes it once a person replies', async () => {
+    await Task.create({
+      podId, taskNum: 1, taskId: 'TASK-1', title: 'Do the assigned work', assignee: 'forge',
+    });
+    for (let i = 0; i < 3; i += 1) await say(agent); // eslint-disable-line no-await-in-loop
+    expect(await wakes(0)).toBe(0);
+    await say(person);
+    expect(await wakes(2)).toBe(1);
+  });
+
+  it('re-arms a row the agent created for itself only when a person posts', async () => {
+    await Task.create({
+      podId, taskNum: 4, taskId: 'N-4-codex-mvid77qk', title: 'Keep posting updates', assignee: 'forge', source: 'agent',
+    });
+    expect(await wakes(0)).toBe(1);
+    await settle(1);
+    await say(agent);
+    expect(await wakes(31)).toBe(0);
+    await say(person);
+    expect(await wakes(62)).toBe(1);
+    await settle(63);
+    expect(await wakes(93)).toBe(0);
+  });
+});