alimek · Oct 11, 2026
diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts
@@ -1,6 +1,7 @@
 import { assert, it } from "@effect/vitest";
 import * as NodeServices from "@effect/platform-node/NodeServices";
-import { PullRequestOperationError } from "@t3tools/contracts";
+import { EnvironmentAuthorizationError, PullRequestOperationError } from "@t3tools/contracts";
+import { trustedReadAuthority } from "../auth/ResourceAuthorization.ts";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Duration from "effect/Duration";
@@ -333,4 +334,95 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
         assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
       }).pipe(Effect.provide(KeyValueStore.layerMemory)),
   );
+  it.effect(
+    "external memory reads retain coalescing, freshness, epochs and cancellation without disk effects",
+    () =>
+      Effect.gen(function* () {
+        const backing = yield* KeyValueStore.KeyValueStore;
+        const effects: string[] = [];
+        const cache = yield* PullRequestReadCache.make.pipe(
+          Effect.provideService(KeyValueStore.KeyValueStore, {
+            ...backing,
+            get: (key) =>
+              Effect.sync(() => {
+                effects.push("get");
+              }).pipe(Effect.andThen(backing.get(key))),
+            set: (key, value) =>
+              Effect.sync(() => {
+                effects.push("set");
+              }).pipe(Effect.andThen(backing.set(key, value))),
+            remove: (key) =>
+              Effect.sync(() => {
+                effects.push("remove");
+              }).pipe(Effect.andThen(backing.remove(key))),
+          }),
+        );
+        const options = {
+          authority: trustedReadAuthority as typeof trustedReadAuthority,
+          revision: "0",
+          // Real admission awaits SQL/canonical paths before lookup; yield once so the cache
+          // awaiter is attached before this synchronous fake lookup announces its start.
+          beforeEffect: Effect.yieldNow,
+        };
+        let reads = 0;
+        const lookup = Effect.sync(() => String(++reads));
+        assert.deepEqual(
+          yield* Effect.all(
+            Array.from({ length: 10 }, () => cache.get("pr", lookup, [], options)),
+            { concurrency: 10 },
+          ),
+          Array(10).fill("1"),
+        );
+        yield* TestClock.adjust("59 seconds");
+        assert.equal(yield* cache.get("pr", lookup, [], options), "1");
+        yield* TestClock.adjust("1 second");
+        assert.equal(yield* cache.get("pr", lookup, [], options), "2");
+        assert.equal(yield* cache.get("pr", lookup, [], { ...options, revision: "1" }), "3");
+        const started = yield* Deferred.make<void>();
+        const read = yield* cache
+          .get(
+            "cancel",
+            Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
+            [],
+            options,
+          )
+          .pipe(Effect.forkChild);
+        yield* Deferred.await(started);
+        yield* Fiber.interrupt(read);
+        assert.equal(yield* cache.get("cancel", lookup, [], options), "4");
+        const denial = new EnvironmentAuthorizationError({
+          requiredScope: "orchestration:read",
+          message: "Denied",
+        });
+        assert.strictEqual(
+          yield* cache.get("denied", Effect.fail(denial), [], options).pipe(Effect.flip),
+          denial,
+        );
+        assert.equal(yield* cache.get("denied", lookup, [], options), "5");
+        assert.deepEqual(effects, []);
+      }).pipe(Effect.provide(KeyValueStore.layerMemory)),
+  );
+
+  it.effect("ordinary revision write failures retain the existing uncached fallback", () =>
+    Effect.gen(function* () {
+      const backing = yield* KeyValueStore.KeyValueStore;
+      const cache = yield* PullRequestReadCache.make.pipe(
+        Effect.provideService(KeyValueStore.KeyValueStore, {
+          ...backing,
+          set: () =>
+            Effect.fail(
+              new KeyValueStore.KeyValueStoreError({ method: "set", message: "unavailable" }),
+            ),
+        }),
+      );
+      yield* cache.invalidate("pr");
+      let reads = 0;
+      const read = cache.get(
+        "summary",
+        Effect.sync(() => String(++reads)),
+      );
+      assert.equal(yield* read, "1");
+      assert.equal(yield* read, "2");
+    }).pipe(Effect.provide(KeyValueStore.layerMemory)),
+  );
 });
diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts
@@ -25,6 +25,7 @@ import * as Persistable from "effect/persistence/Persistable";
 import * as PersistedCache from "effect/persistence/PersistedCache";
 import * as Persistence from "effect/persistence/Persistence";
 import * as ServerConfig from "../config.ts";
+import type { ThreadReadAuthority } from "../auth/ResourceAuthorization.ts";
 import { forkParked } from "../serverActivation.ts";
 
 const CONCURRENT_READS = 512;
@@ -43,14 +44,25 @@ type ReadError =
   | PullRequestOperationError
   | PullRequestUnavailableError
   | EnvironmentAuthorizationError;
+const isAuthorizationError = Schema.is(EnvironmentAuthorizationError);
+const beforeRevisionEffect = Context.Reference<Effect.Effect<void, EnvironmentAuthorizationError>>(
+  "t3/pullRequest/readCacheAdmission",
+  { defaultValue: () => Effect.void },
+);
+const checkRevisionAccess = beforeRevisionEffect.pipe(Effect.flatten);
 const revisionCodec = Schema.fromJsonString(
   Schema.Record(
     Schema.String,
     Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }),
   ),
 );
 class Read extends Persistable.Class<{
-  payload: { key: string; revision: string; lookup: Effect.Effect<string, ReadError> };
+  payload: {
+    key: string;
+    revision: string;
+    lookup: Effect.Effect<string, ReadError>;
+    authority?: ThreadReadAuthority;
+  };
 }>()("PullRequestRead", {
   primaryKey: ({ key }) => key,
   success: Schema.Struct({
@@ -72,10 +84,15 @@ class Read extends Persistable.Class<{
   }
 
   [Equal.symbol](that: unknown): boolean {
-    return that instanceof Read && that.key === this.key && that.revision === this.revision;
+    return (
+      that instanceof Read &&
+      that.key === this.key &&
+      that.revision === this.revision &&
+      that.authority === this.authority
+    );
   }
   [Hash.symbol](): number {
-    return Hash.string(`${this.key}:${this.revision}`);
+    return Hash.combine(Hash.hash(this.authority))(Hash.string(`${this.key}:${this.revision}`));
   }
 }
 
@@ -86,8 +103,16 @@ export class PullRequestReadCache extends Context.Service<
       key: string,
       lookup: Effect.Effect<string, ReadError>,
       scopes?: ReadonlyArray<string>,
+      external?: {
+        readonly authority: ThreadReadAuthority;
+        readonly revision: string;
+        readonly beforeEffect: Effect.Effect<void, EnvironmentAuthorizationError>;
+      },
     ) => Effect.Effect<string, ReadError>;
-    readonly invalidate: (scope: string) => Effect.Effect<void>;
+    readonly invalidate: (
+      scope: string,
+      beforeEffect?: Effect.Effect<void, EnvironmentAuthorizationError>,
+    ) => Effect.Effect<void, EnvironmentAuthorizationError>;
   }
 >()("t3/pullRequest/PullRequestReadCache") {}
 
@@ -101,9 +126,13 @@ export const make = Effect.gen(function* () {
     crypto.digest("SHA-256", new TextEncoder().encode(key)).pipe(Effect.map(Hex.encode));
   const revisions = yield* Cache.makeWith(
     () =>
-      backing
-        .get("revisions")
-        .pipe(Effect.flatMap((raw) => Schema.decodeEffect(revisionCodec)(raw ?? "{}"))),
+      checkRevisionAccess.pipe(
+        Effect.andThen(backing.get("revisions")),
+        Effect.flatMap((raw) =>
+          checkRevisionAccess.pipe(Effect.andThen(Schema.decodeEffect(revisionCodec)(raw ?? "{}"))),
+        ),
+        Effect.tap(() => checkRevisionAccess),
+      ),
     {
       capacity: 1,
       timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
@@ -113,22 +142,26 @@ export const make = Effect.gen(function* () {
     Exit.isSuccess(exit)
       ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
       : Duration.zero;
-  const cache = yield* PersistedCache.make(
-    (request: Read) =>
-      request.lookup.pipe(
-        Effect.map((payload) => ({
-          payload,
-          expiresAt: clock.currentTimeMillisUnsafe() + 60_000,
-          revision: request.revision,
-        })),
-      ),
-    {
-      storeId: STORE_ID,
-      timeToLive,
-      inMemoryTTL: timeToLive,
-      inMemoryCapacity: CONCURRENT_READS,
-    },
-  ).pipe(Effect.provide(Persistence.layerKvs));
+  const lookupRead = (request: Read) =>
+    request.lookup.pipe(
+      Effect.map((payload) => ({
+        payload,
+        expiresAt: clock.currentTimeMillisUnsafe() + 60_000,
+        revision: request.revision,
+      })),
+    );
+  // External reads use the service's epochs and stay off disk. PersistedCache writes after
+  // lookup/serialization and its KVS layer can swallow an expired-entry removal failure.
+  const externalReads = yield* Cache.makeWith(lookupRead, {
+    capacity: CONCURRENT_READS,
+    timeToLive,
+  });
+  const cache = yield* PersistedCache.make(lookupRead, {
+    storeId: STORE_ID,
+    timeToLive,
+    inMemoryTTL: timeToLive,
+    inMemoryCapacity: CONCURRENT_READS,
+  }).pipe(Effect.provide(Persistence.layerKvs));
   const refreshes = yield* Cache.makeWith(
     Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) {
       const stored = yield* cache.get(request);
@@ -139,7 +172,24 @@ export const make = Effect.gen(function* () {
     { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
   );
   return PullRequestReadCache.of({
-    get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
+    get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = [], external) {
+      if (external !== undefined) {
+        return yield* Effect.gen(function* () {
+          yield* external.beforeEffect;
+          const request = new Read({
+            key,
+            revision: external.revision,
+            authority: external.authority,
+            lookup: external.beforeEffect.pipe(
+              Effect.andThen(lookup),
+              Effect.tap(() => external.beforeEffect),
+            ),
+          });
+          const stored = yield* Cache.get(externalReads, request);
+          yield* external.beforeEffect;
+          return stored.payload;
+        }).pipe(lock.withPermits(1));
+      }
       if (!enabled) return yield* lookup;
       const read = yield* Effect.cached(lookup);
       return yield* Effect.gen(function* () {
@@ -166,22 +216,39 @@ export const make = Effect.gen(function* () {
         lock.withPermits(1),
       );
     }),
-    invalidate: (scope) =>
+    invalidate: (scope, beforeEffect = Effect.void) =>
       Effect.gen(function* () {
+        yield* beforeEffect;
         const now = clock.currentTimeMillisUnsafe();
         const current = yield* Cache.get(revisions, undefined);
+        yield* beforeEffect;
         const next = Object.fromEntries(
           Object.entries(current).filter(([, value]) => value.expiresAt > now),
         );
         next[scope] = { revision: yield* crypto.randomUUIDv4, expiresAt: now + 60_000 };
+        yield* beforeEffect;
         const encoded = yield* Schema.encodeEffect(revisionCodec)(next);
+        yield* beforeEffect;
+        // If authority changes while the write is in flight, the next admitted caller reloads
+        // disk instead of reusing the revision that preceded this write.
+        yield* Cache.invalidate(revisions, undefined);
+        yield* beforeEffect;
         yield* backing.set("revisions", encoded);
+        yield* beforeEffect;
         yield* Cache.set(revisions, undefined, next);
       }).pipe(
-        Effect.catch(() => {
-          enabled = false;
-          return Effect.logWarning("PR cache disabled after clearing failed");
+        Effect.catch((error) => {
+          if (isAuthorizationError(error)) return Effect.fail(error);
+          return beforeEffect.pipe(
+            Effect.andThen(
+              Effect.sync(() => {
+                enabled = false;
+              }),
+            ),
+            Effect.andThen(Effect.logWarning("PR cache disabled after clearing failed")),
+          );
         }),
+        Effect.provideService(beforeRevisionEffect, beforeEffect),
         Effect.uninterruptible,
         lock.withPermits(CONCURRENT_READS),
       ),