fix(runtime): cache unresolved historical results

Author: me2seeks

packages/runtime/src/__tests__/stream-graph-schedule-reconcile.test.ts

@@ -113,6 +113,7 @@ describe('stream graph schedule reconciliation', () => {
           },
         ],
       });
+      let resolveCalls = 0;
       const result = await reconcileAgentGraphSchedule({
         topology: topology(),
         controlStore,
@@ -122,6 +123,7 @@ describe('stream graph schedule reconciliation', () => {
         maxNewActivations: 1,
         observeGraph: (currentTopology) => observation.read(currentTopology),
         resolveSelectedResultInputs: async () => {
+          resolveCalls += 1;
           throw new Error('source epoch runtime events are unreadable');
         },
         renderPrompt: ({ work }) => work.instruction,
@@ -130,6 +132,7 @@ describe('stream graph schedule reconciliation', () => {
       assert.equal(result.status, 'waiting');
       assert.equal(result.failures.length, 0);
       assert.equal(result.dispatches.length, 0);
+      assert.equal(resolveCalls, 1);
       assert.deepEqual(
         result.deferredWork.map((item) => ({
           reason: item.reason,

packages/runtime/src/stream-graph-schedule-reconcile.ts

@@ -513,8 +513,8 @@ function missingWorkInputIds(
   ];
 }
 
-/** Resolved historical records keyed by source graph id, then result id. */
-type SelectedResultCache = Map<string, Map<string, AgentGraphRecord>>;
+/** Historical resolution attempts keyed by source graph id, then result id. */
+type SelectedResultCache = Map<string, Map<string, AgentGraphRecord | undefined>>;
 
 /**
  * Resolves selected historical result inputs, isolating failures per source
@@ -557,6 +557,10 @@ async function resolveSelectedResultRecords(
     }
     const pending = group.filter((selected) => !cached.has(selected.resultId));
     if (pending.length === 0) continue;
+    // Cache unsuccessful attempts for this reconciliation too. Repeated
+    // snapshot reads must not replay a resolver that already failed or
+    // returned records that violate its identity contract.
+    pending.forEach((selected) => cached.set(selected.resultId, undefined));
     let records: readonly AgentGraphRecord[];
     try {
       records = await input.resolveSelectedResultInputs(pending);