fix(agenstra): new message marker persistence

libs/domains/agenstra/backend/feature-agent-controller/src/lib/repositories/user-chat-session-read-state.repository.ts

@@ -4,6 +4,14 @@ import { In, Repository } from 'typeorm';
 
 import { UserChatSessionReadStateEntity } from '../entities/user-chat-session-read-state.entity';
 
+function laterDate(a: Date | null | undefined, b: Date): Date {
+  if (!a) {
+    return b;
+  }
+
+  return a.getTime() > b.getTime() ? a : b;
+}
+
 @Injectable()
 export class UserChatSessionReadStateRepository {
   constructor(
@@ -20,6 +28,11 @@ export class UserChatSessionReadStateRepository {
     return await this.repository.findOne({ where: { userId, clientId, agentId, chatSessionId } });
   }
 
+  /**
+   * Upsert read cursor. Never moves `lastReadAt` backwards.
+   * Omit `lastReadAgentMessageId` (undefined) to keep the existing message id;
+   * pass null only when intentionally clearing.
+   */
   async upsertReadState(params: {
     userId: string;
     clientId: string;
@@ -31,8 +44,11 @@ export class UserChatSessionReadStateRepository {
     const existing = await this.findOne(params.userId, params.clientId, params.agentId, params.chatSessionId);
 
     if (existing) {
-      existing.lastReadAt = params.lastReadAt;
-      existing.lastReadAgentMessageId = params.lastReadAgentMessageId ?? null;
+      existing.lastReadAt = laterDate(existing.lastReadAt, params.lastReadAt);
+
+      if (params.lastReadAgentMessageId !== undefined) {
+        existing.lastReadAgentMessageId = params.lastReadAgentMessageId;
+      }
 
       return await this.repository.save(existing);
     }

libs/domains/agenstra/backend/feature-agent-controller/src/lib/repositories/user-environment-read-state.repository.spec.ts

@@ -80,6 +80,27 @@ describe('UserEnvironmentReadStateRepository', () => {
       expect(mockTypeOrmRepository.create).not.toHaveBeenCalled();
     });
 
+    it('does not move lastReadAt backwards or clear message id when omitted', async () => {
+      const existing = {
+        ...mockRow,
+        lastReadAt: new Date('2026-02-01T00:00:00.000Z'),
+        lastReadAgentMessageId: 'msg-keep',
+      };
+
+      mockTypeOrmRepository.findOne.mockResolvedValue(existing);
+      mockTypeOrmRepository.save.mockImplementation(async (row) => row);
+
+      const result = await repository.upsertReadState({
+        userId: 'user-1',
+        clientId: 'client-1',
+        agentId: 'agent-1',
+        lastReadAt: new Date('2026-01-01T00:00:00.000Z'),
+      });
+
+      expect(result.lastReadAt).toEqual(new Date('2026-02-01T00:00:00.000Z'));
+      expect(result.lastReadAgentMessageId).toBe('msg-keep');
+    });
+
     it('creates row when missing', async () => {
       mockTypeOrmRepository.findOne.mockResolvedValue(null);
       mockTypeOrmRepository.create.mockReturnValue(mockRow);

libs/domains/agenstra/backend/feature-agent-controller/src/lib/repositories/user-environment-read-state.repository.ts

@@ -29,8 +29,13 @@ export class UserEnvironmentReadStateRepository {
     const existing = await this.findOne(params.userId, params.clientId, params.agentId);
 
     if (existing) {
-      existing.lastReadAt = params.lastReadAt;
-      existing.lastReadAgentMessageId = params.lastReadAgentMessageId ?? null;
+      const existingAt = existing.lastReadAt;
+      existing.lastReadAt =
+        existingAt && existingAt.getTime() > params.lastReadAt.getTime() ? existingAt : params.lastReadAt;
+
+      if (params.lastReadAgentMessageId !== undefined) {
+        existing.lastReadAgentMessageId = params.lastReadAgentMessageId;
+      }
 
       return await this.repository.save(existing);
     }

libs/domains/agenstra/backend/feature-agent-controller/src/lib/services/agent-console-status.service.spec.ts

@@ -212,6 +212,62 @@ describe('AgentConsoleStatusService', () => {
     expect(realtime.emitToUser).toHaveBeenCalledWith('user-1', 'statusPatch', expect.any(Object));
   });
 
+  it('computes unread after VCS so a mark-read during build is visible', async () => {
+    const messageCreatedAt = new Date('2026-01-02T00:00:00.000Z');
+    messagesProxy.getLatestAgentMessage.mockResolvedValue({
+      id: 'msg-1',
+      createdAt: messageCreatedAt.toISOString(),
+    });
+
+    let vcsResolve: ((value: unknown) => void) | undefined;
+    const vcsGate = new Promise((resolve) => {
+      vcsResolve = resolve;
+    });
+
+    vcsProxy.getStatus.mockImplementation(async () => {
+      await vcsGate;
+
+      return {
+        isClean: true,
+        hasUnpushedCommits: false,
+        files: [],
+      };
+    });
+
+    const snapshotPromise = service.buildSnapshotForUser({
+      isApiKeyAuth: false,
+      userId: 'user-1',
+      userRole: UserRole.USER,
+      user: { id: 'user-1', roles: [] },
+    });
+
+    // Let phase-1 reach the gated VCS call.
+    await Promise.resolve();
+    await Promise.resolve();
+
+    // Mark-read lands while VCS is still in flight; phase-2 must observe it.
+    chatReadStateRepository.findByUserAndClientIds.mockResolvedValue([
+      {
+        userId: 'user-1',
+        clientId: 'client-1',
+        agentId: 'agent-1',
+        chatSessionId: USER_CHAT_ID,
+        lastReadAt: new Date('2026-01-03T00:00:00.000Z'),
+      },
+    ]);
+
+    vcsResolve?.({
+      isClean: true,
+      hasUnpushedCommits: false,
+      files: [],
+    });
+
+    const snapshot = await snapshotPromise;
+    const chats = snapshot.environments[0].chats ?? [];
+
+    expect(chats.find((c) => c.chatSessionId === USER_CHAT_ID)?.hasUnreadMessages).toBe(false);
+  });
+
   it('notifyVcsStateChanged emits status patches to users with client access', async () => {
     vcsProxy.getStatus.mockResolvedValue({
       isClean: true,

libs/domains/agenstra/backend/feature-agent-controller/src/lib/services/agent-console-status.service.ts

@@ -152,12 +152,14 @@ export class AgentConsoleStatusService {
 
     const resolvedChatId = chatSessionId ?? (await this.resolvePrimaryChatId(clientId, agentId));
     const latest = await this.messagesProxy.getLatestAgentMessage(clientId, agentId, resolvedChatId ?? undefined);
+    const latestAt = latest?.createdAt ? new Date(latest.createdAt) : null;
+    const readAt = this.maxDate(new Date(), latestAt) ?? new Date();
 
     await this.readStateRepository.upsertReadState({
       userId,
       clientId,
       agentId,
-      lastReadAt: new Date(),
+      lastReadAt: readAt,
       lastReadAgentMessageId: latest?.id ?? null,
     });
 
@@ -167,7 +169,7 @@ export class AgentConsoleStatusService {
         clientId,
         agentId,
         chatSessionId: resolvedChatId,
-        lastReadAt: new Date(),
+        lastReadAt: readAt,
         lastReadAgentMessageId: latest?.id ?? null,
       });
     }
@@ -190,21 +192,23 @@ export class AgentConsoleStatusService {
     await this.assertEnvironmentAccess(userInfo, clientId, agentId);
 
     const latest = await this.messagesProxy.getLatestAgentMessage(clientId, agentId, chatSessionId);
+    const latestAt = latest?.createdAt ? new Date(latest.createdAt) : null;
+    const readAt = this.maxDate(new Date(), latestAt) ?? new Date();
 
     await this.chatReadStateRepository.upsertReadState({
       userId,
       clientId,
       agentId,
       chatSessionId,
-      lastReadAt: new Date(),
+      lastReadAt: readAt,
       lastReadAgentMessageId: latest?.id ?? null,
     });
 
     await this.readStateRepository.upsertReadState({
       userId,
       clientId,
       agentId,
-      lastReadAt: new Date(),
+      lastReadAt: readAt,
       lastReadAgentMessageId: latest?.id ?? null,
     });
 
@@ -302,6 +306,11 @@ export class AgentConsoleStatusService {
     activityAt: Date,
   ): Promise<void> {
     const userIds = await this.resolveUserIdsToNotify(clientId);
+    const latest = chatSessionId
+      ? await this.messagesProxy.getLatestAgentMessage(clientId, agentId, chatSessionId)
+      : await this.messagesProxy.getLatestAgentMessage(clientId, agentId);
+    const latestAt = latest?.createdAt ? new Date(latest.createdAt) : null;
+    const readAt = this.maxDate(activityAt, latestAt) ?? activityAt;
 
     for (const userId of userIds) {
       const activeForUser = this.findActiveEnvironmentForUser(userId, clientId, agentId, chatSessionId);
@@ -311,8 +320,9 @@ export class AgentConsoleStatusService {
           userId,
           clientId,
           agentId,
-          lastReadAt: activityAt,
-          lastReadAgentMessageId: null,
+          lastReadAt: readAt,
+          // Keep prior message id when latest is unknown — never wipe a successful mark-read.
+          ...(latest?.id ? { lastReadAgentMessageId: latest.id } : {}),
         });
 
         if (chatSessionId) {
@@ -321,8 +331,8 @@ export class AgentConsoleStatusService {
             clientId,
             agentId,
             chatSessionId,
-            lastReadAt: activityAt,
-            lastReadAgentMessageId: null,
+            lastReadAt: readAt,
+            ...(latest?.id ? { lastReadAgentMessageId: latest.id } : {}),
           });
         }
       }
@@ -401,16 +411,21 @@ export class AgentConsoleStatusService {
   }
 
   private async buildEnvironmentsForUser(userId: string, clientIds: string[]): Promise<EnvironmentStatusPayload[]> {
-    const chatReadStates = await this.chatReadStateRepository.findByUserAndClientIds(userId, clientIds);
-    const chatReadByKey = new Map<string, (typeof chatReadStates)[0]>();
-
-    for (const row of chatReadStates) {
-      chatReadByKey.set(`${row.clientId}:${row.agentId}:${row.chatSessionId}`, row);
-    }
+    type PendingEnv = {
+      clientId: string;
+      agentId: string;
+      gitDirty: boolean;
+      gitConflict: boolean;
+      visibleChats: Array<{ id: string; kind: string }>;
+      primaryChatId: string | null;
+      latestAutomationAt: Date | null;
+    };
 
-    const environments: EnvironmentStatusPayload[] = [];
+    const pending: PendingEnv[] = [];
     const vcsConcurrency = this.getVcsConcurrency();
 
+    // Phase 1: resolve agents + slow VCS. Unread is computed afterward so a mark-read
+    // that lands during VCS cannot be overwritten by a stale pre-mark read snapshot.
     for (const clientId of clientIds) {
       let agents: AgentResponseDto[] = [];
 
@@ -428,30 +443,11 @@ export class AgentConsoleStatusService {
         const chunk = agents.slice(i, i + vcsConcurrency);
         const chunkResults = await Promise.all(
           chunk.map(async (agent) => {
-            const visibleChats = (agent.chats ?? []).filter((c) => c.kind === 'primary' || c.kind === 'user');
+            const visibleChats = (agent.chats ?? [])
+              .filter((c) => c.kind === 'primary' || c.kind === 'user')
+              .map((c) => ({ id: c.id, kind: c.kind }));
             const primaryChatId = agent.primaryChatId ?? visibleChats.find((c) => c.kind === 'primary')?.id ?? null;
             const latestAutomationAt = automationByAgent.get(agent.id) ?? null;
-
-            const chatStatuses: ChatSessionStatusPayload[] = await Promise.all(
-              visibleChats.map(async (chat) => {
-                const latestAgentMsg = await this.messagesProxy.getLatestAgentMessage(clientId, agent.id, chat.id);
-                const latestAgentAt = latestAgentMsg ? new Date(latestAgentMsg.createdAt) : null;
-                const activityAt =
-                  chat.kind === 'primary' || chat.id === primaryChatId
-                    ? this.maxDate(latestAgentAt, latestAutomationAt)
-                    : latestAgentAt;
-                const readState = chatReadByKey.get(`${clientId}:${agent.id}:${chat.id}`);
-                const lastReadAt = readState?.lastReadAt ?? null;
-                const hasUnreadMessages = activityAt !== null && (lastReadAt === null || activityAt > lastReadAt);
-
-                return {
-                  chatSessionId: chat.id,
-                  hasUnreadMessages,
-                };
-              }),
-            );
-
-            const hasUnreadMessages = chatStatuses.some((c) => c.hasUnreadMessages);
             let gitDirty = false;
             let gitConflict = false;
 
@@ -468,19 +464,72 @@ export class AgentConsoleStatusService {
             return {
               clientId,
               agentId: agent.id,
-              hasUnreadMessages,
               gitDirty,
               gitConflict,
-              chats: chatStatuses,
-            } satisfies EnvironmentStatusPayload;
+              visibleChats,
+              primaryChatId,
+              latestAutomationAt,
+            } satisfies PendingEnv;
           }),
         );
 
-        environments.push(...chunkResults);
+        pending.push(...chunkResults);
       }
     }
 
-    return environments;
+    // Phase 2: fresh read cursors + latest messages (after any concurrent mark-read).
+    const chatReadStates = await this.chatReadStateRepository.findByUserAndClientIds(userId, clientIds);
+    const chatReadByKey = new Map<string, (typeof chatReadStates)[0]>();
+
+    for (const row of chatReadStates) {
+      chatReadByKey.set(`${row.clientId}:${row.agentId}:${row.chatSessionId}`, row);
+    }
+
+    return await Promise.all(
+      pending.map(async (item) => {
+        const chatStatuses: ChatSessionStatusPayload[] = await Promise.all(
+          item.visibleChats.map(async (chat) => {
+            const latestAgentMsg = await this.messagesProxy.getLatestAgentMessage(item.clientId, item.agentId, chat.id);
+            const latestAgentAt = latestAgentMsg ? new Date(latestAgentMsg.createdAt) : null;
+            const isPrimary = chat.kind === 'primary' || chat.id === item.primaryChatId;
+            const readState = chatReadByKey.get(`${item.clientId}:${item.agentId}:${chat.id}`);
+            const lastReadAt = readState?.lastReadAt ?? null;
+            const lastReadMessageId = readState?.lastReadAgentMessageId ?? null;
+
+            // Prefer message-id equality so clock skew / auto-clear races cannot revive unread.
+            let messageUnread = false;
+
+            if (latestAgentMsg) {
+              if (lastReadMessageId && lastReadMessageId === latestAgentMsg.id) {
+                messageUnread = false;
+              } else {
+                messageUnread = lastReadAt === null || (latestAgentAt !== null && latestAgentAt > lastReadAt);
+              }
+            }
+
+            const automationUnread =
+              isPrimary &&
+              item.latestAutomationAt !== null &&
+              (lastReadAt === null || item.latestAutomationAt > lastReadAt);
+            const hasUnreadMessages = messageUnread || Boolean(automationUnread);
+
+            return {
+              chatSessionId: chat.id,
+              hasUnreadMessages,
+            };
+          }),
+        );
+
+        return {
+          clientId: item.clientId,
+          agentId: item.agentId,
+          hasUnreadMessages: chatStatuses.some((c) => c.hasUnreadMessages),
+          gitDirty: item.gitDirty,
+          gitConflict: item.gitConflict,
+          chats: chatStatuses,
+        } satisfies EnvironmentStatusPayload;
+      }),
+    );
   }
 
   private async buildEnvironmentStatus(