fix(agent-graph): validate historical epoch queries

Author: me2seeks

packages/runtime-host/src/__tests__/agent-graph-coordinator.test.ts

@@ -252,6 +252,13 @@ describe('Host Agent Graph coordinator', () => {
       ],
     );
     assert.equal(second.result.nextBeforeEpoch, null);
+
+    const ahead = await coordinator.handlers['agent.graph.epochs.query'](
+      { rootSessionId: 'root-1', beforeEpoch: 999 },
+      context(),
+    );
+    assert.equal(ahead.ok, false);
+    if (!ahead.ok) assert.equal(ahead.error.code, 'invalid_request');
     coordinator.close();
   });
 

packages/runtime-host/src/server/agent-graph-coordinator.ts

@@ -137,6 +137,12 @@ export class HostAgentGraphCoordinator {
         ...(input.beforeEpoch === undefined ? {} : { beforeEpoch: input.beforeEpoch }),
         limit: AGENT_GRAPH_EPOCH_PAGE_SIZE,
       });
+      if (input.beforeEpoch !== undefined && input.beforeEpoch > page.currentEpoch) {
+        throw new AgentGraphClientOperationError(
+          'invalid_request',
+          `Agent graph epoch cursor ${input.beforeEpoch} is ahead of current epoch ${page.currentEpoch}`,
+        );
+      }
       return {
         ok: true,
         result: {

packages/runtime/src/__tests__/stream-graph-coordinator.test.ts

@@ -19,6 +19,7 @@ import { type RuntimeEvent } from '@maka/core/runtime-event';
 import {
   createSessionStore,
   createSqliteSessionMetadataStore,
+  isSessionNotFoundError,
   OPERATIONAL_STATE_DATABASE_NAME,
 } from '@maka/storage';
 import { createSqliteAgentRunStore, createWorkspaceRuntimeStore } from '@maka/storage';
@@ -328,6 +329,15 @@ describe('host-managed agent graph coordinator', () => {
         coordinator.toolsForSession(childSessions[0]!.id),
         /only to root Sessions/,
       );
+      await assert.rejects(
+        coordinator.listGraphEpochPage(childSessions[0]!.id, { limit: 32 }),
+        (error: unknown) =>
+          error instanceof AgentGraphClientOperationError && error.code === 'operation_conflict',
+      );
+      await assert.rejects(
+        coordinator.listGraphEpochPage(randomUUID(), { limit: 32 }),
+        (error: unknown) => isSessionNotFoundError(error),
+      );
       let childStopEntered = false;
       await assert.rejects(
         coordinator.stopExecution(childSessions[0]!.id, {
@@ -545,11 +555,18 @@ describe('host-managed agent graph coordinator', () => {
   test('reads the epoch page and current marker from one storage observation', async () => {
     const controlStore = createSqliteSessionMetadataStore(':memory:');
     let resolveCalls = 0;
+    let headerReads = 0;
     const coordinator = new AgentGraphCoordinator({
       sessionStore: {
         listForRecovery: async () => [],
-        readHeader: async () => {
-          throw new Error('epoch listing must not read the Session header');
+        readHeader: async (sessionId: string) => {
+          headerReads += 1;
+          return {
+            id: sessionId,
+            status: 'active',
+            isArchived: false,
+            orchestrationMode: 'graph',
+          } as never;
         },
       },
       runStore: { listSessionRuns: async () => [] },
@@ -612,6 +629,12 @@ describe('host-managed agent graph coordinator', () => {
       assert.equal(page.currentEpoch, 3);
       assert.equal(page.epochs[0]?.graphId, 'agent_graph_3');
       assert.equal(resolveCalls, 0);
+      assert.equal(headerReads, 1);
+      await assert.rejects(
+        coordinator.listGraphEpochPage('root-session', { beforeEpoch: 999, limit: 32 }),
+        (error: unknown) =>
+          error instanceof AgentGraphClientOperationError && error.code === 'invalid_request',
+      );
     } finally {
       await coordinator.close();
       controlStore.close();

packages/runtime/src/stream-graph-coordinator.ts

@@ -294,7 +294,7 @@ export class AgentGraphCoordinator {
   }
 
   async listGraphEpochs(rootSessionId: string): Promise<readonly AgentGraphEpochBinding[]> {
-    requireRootSessionId(rootSessionId);
+    await this.#assertRootGraphReader(rootSessionId);
     const current = await this.currentGraphEpoch(rootSessionId);
     if (!this.#input.epochStore) return [current];
     const epochs = await this.#input.epochStore.listAgentGraphEpochs(rootSessionId);
@@ -305,9 +305,10 @@ export class AgentGraphCoordinator {
     rootSessionId: string,
     options: { readonly beforeEpoch?: number; readonly limit: number },
   ): Promise<AgentGraphEpochPage & { readonly currentEpoch: number }> {
-    requireRootSessionId(rootSessionId);
+    await this.#assertRootGraphReader(rootSessionId);
     if (!this.#input.epochStore) {
       const current = await this.currentGraphEpoch(rootSessionId);
+      assertEpochCursorNotAhead(options.beforeEpoch, current.epoch);
       return {
         epochs:
           options.beforeEpoch === undefined || current.epoch < options.beforeEpoch ? [current] : [],
@@ -323,6 +324,7 @@ export class AgentGraphCoordinator {
       ...options,
     });
     if (page.currentEpoch !== null) {
+      assertEpochCursorNotAhead(options.beforeEpoch, page.currentEpoch);
       return {
         epochs: page.epochs,
         nextBeforeEpoch: page.nextBeforeEpoch,
@@ -331,6 +333,7 @@ export class AgentGraphCoordinator {
     }
     // No durable rows yet: synthesize the legacy virtual epoch identity.
     const current = await this.currentGraphEpoch(rootSessionId);
+    assertEpochCursorNotAhead(options.beforeEpoch, current.epoch);
     return {
       epochs: options.beforeEpoch === undefined ? [current] : [],
       nextBeforeEpoch: null,
@@ -1488,6 +1491,15 @@ export class AgentGraphCoordinator {
   }
 }
 
+function assertEpochCursorNotAhead(beforeEpoch: number | undefined, currentEpoch: number): void {
+  if (beforeEpoch !== undefined && beforeEpoch > currentEpoch) {
+    throw new AgentGraphClientOperationError(
+      'invalid_request',
+      `Agent graph epoch cursor ${beforeEpoch} is ahead of current epoch ${currentEpoch}`,
+    );
+  }
+}
+
 interface ScheduleWakeFence {
   stopGeneration: number;
   mayResumePaused: boolean;