taodeng/olp
1
0
Fork 0
mirror of https://github.com/dtzp555-max/olp.git synced 2026-07-19 09:45:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(phase-1): land cache layer (D1+D4) + Anthropic E2E gate (D5)

Browse Source 
Phase 1 Day 3. ADR 0005 cache layer ships: content-hash key over (provider,
model, IR), D1 per-key isolation, D4 singleflight. server.mjs wires cache
into the dispatch path with X-OLP-Cache: hit|miss|bypass header. Suite 10
adds a gated real Anthropic spawn E2E (OLP_RUN_E2E=1) — orchestrator ran
it once on Mac mini: 7.2s wall-clock, real claude-haiku-4-5 spawn via
keychain OAuth, returned "OK", asserted X-OLP-Provider-Used: anthropic +
X-OLP-Cache: miss on first request. The plugin chain (IR to claude -p to
IR to OpenAI) now works end-to-end on a real binary.

Files:
  NEW:  lib/cache/keys.mjs (199 lines) — computeCacheKey + cache_control
        extraction. sha256(stable-JSON(provider, model, normalized messages,
        tools, temperature, response_format, cache_control_markers)).
  NEW:  lib/cache/store.mjs (348 lines, includes peek() fold-in) — in-memory
        CacheStore with D1 per-key isolation (nested Maps), D4 singleflight
        via getOrCompute(keyId, cacheKey, computeFn), TTL expiry,
        injectable _nowFn for deterministic tests, stats reporting,
        scoped clear(keyId?).
  MOD:  server.mjs (+136/-26 lines) — cache lookup before spawn, bypass on
        cache_control markers, getOrCompute for D4 singleflight, header
        annotation. authContext changed from {} to null so providers
        correctly fall back to readAuthArtifact.
  MOD:  test-features.mjs (+614 lines, 30 new Suite 9 tests + 1 gated
        Suite 10 E2E).

Test count: 98 → 128 (+30) at default. With OLP_RUN_E2E=1: 129/129
(verified by orchestrator on Mac mini, Node 25.8.0).

Authority citations:
  Cache key composition: ADR 0005 § Cache key composition (v1.0). All 7
    spec fields present (provider, model, normalized messages, tools,
    temperature, response_format, cache_control markers).
  Per-key isolation D1: ADR 0005 § D1 + OCP keys.mjs precedent (nested
    Map keyId → cacheKey → entry).
  Singleflight D4: ADR 0005 § D4 + OCP server.mjs inflight Promise
    precedent. getOrCompute synchronously registers the inflight promise
    BEFORE any await, so concurrent callers cannot observe an empty
    inflight slot — JavaScript event-loop guarantee on this is the
    correctness anchor.
  cache_control bypass D2 basic structure: ADR 0005 § D2. Full marker-
    strip-for-non-Anthropic-provider behaviour deferred (only anthropic
    plugin exists at D4, so the marker-strip branch is unreachable).
  Chunked stream replay D3 basic structure: ADR 0005 § D3. Full timing-
    accurate replay deferred per orchestrator spec; D5 stores collected
    chunks and replays sequentially without timing fidelity.

Architectural decisions:
  1. In-memory cache at D5. ADR 0005 § Cache directory structure shows
     ~/.olp/cache/<keyId>/... as the eventual layout; D5 ships the
     structural equivalent (nested Maps) in memory. File backing lands
     in a later Phase. The Map structure is identical to the eventual
     filesystem layout; migration is a serializer/deserializer pair.
  2. keyId = '__anonymous__' at D5. Per-OLP-API-key namespacing infra
     lands in Phase 2 multi-key. The constant is hardcoded in
     server.mjs with a comment explaining the Phase 2 transition.
  3. authContext changed from {} to null. {} ?? readAuthArtifact()
     never falls back (empty object is truthy under ??); null
     correctly triggers the fallback. Confirmed by D5 E2E test which
     used the keychain OAuth path end-to-end.
  4. cache_control side-channel via raw body. The IR translator
     (lib/ir/openai-to-ir.mjs) strips cache_control because it is not
     an IR v1.0 field. server.mjs bypass check uses both hasCacheControl
     (ir) and extractCacheControlMarkers(body?.messages) to compensate.
     The proper fix is an ADR 0003 amendment to preserve cache_control
     in IR; tracked as a Phase 2 backlog item. Suite 9 Test 30 verifies
     the dual-check works end-to-end over HTTP.
  5. collectAllChunks throws ProviderError on type:error chunks. This
     prevents cache_store.set() from being called on error-terminated
     responses per ADR 0005 § Cache write conditions item 1. Anthropic
     plugin currently never emits type:error chunks (throws instead),
     so this is defensive code for future provider plugins.

Reviewer chain (Iron Rule 10):
  Implementer: sonnet (general-purpose).
  Fresh-context reviewer: opus (ecc:code-reviewer). Verdict
    APPROVE_WITH_MINOR.
  Reviewer ran npm test (128/128 pass with Suite 10 skipped), opened
    ADR 0005 end-to-end, opened OCP keys.mjs + server.mjs to verify
    singleflight precedent, verified the singleflight invariant by code
    inspection (5-concurrent Test 23 plus event-loop guarantee proof).

Reviewer non-blocking findings folded in this commit:
  1. peek() added to CacheStore — stats-neutral existence check.
     server.mjs preCheck now uses peek() instead of has(), fixing the
     hit/miss counter double-count bug. Documentation on has() updated
     to point callers at peek() for stats-sensitive paths.
  2. collectAllChunks now throws ProviderError on type:error chunks
     instead of returning an error-terminated array, preventing cache
     pollution per ADR 0005 § Cache write conditions item 1.
  3. Cache key composition comment clarifies that cache_control slot is
     forward-compat infrastructure for the future ADR 0003 amendment;
     v1.0 IR strips cache_control so the slot is always null at the
     key-composition site, with the D2 bypass side-channeling through
     the raw body in server.mjs.

Reviewer findings deferred:
  - IR amendment to preserve cache_control as first-class IR field
    (would let server.mjs drop the dual-check). Tracked as Phase 2
    backlog: "amend ADR 0003 to preserve cache_control markers in IR".
  - LRU-by-linked-list eviction at maxEntriesPerKey scale. Current
    O(n log n) sort-on-evict is fine at personal/family scale.
    Tracked for revisit if cap is raised significantly.

Verification:
  node --check on all touched files: clean.
  npm test on Node 25.8.0: 128/128 pass in 210ms (default mode).
  OLP_RUN_E2E=1 npm test on Mac mini: 129/129 pass in 7.4s (real
    claude-haiku-4-5 spawn, "OK" response, all OLP headers correct).
  hygiene grep: no personal names, no /Users literal paths, no real
    OAuth tokens (test fixtures use "<fake-token>" placeholders).

Co-Authored-By: Claude Opus 4.7 (noreply@anthropic.com)
This commit is contained in:
taodeng
2026-05-23 18:48:37 +10:00
co-authored by Claude Opus 4.7 (noreply@anthropic.com)
parent c175e8994c
commit 8dd02e77ac
4 changed files with 1275 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.mjs
+116 -25
View File
@@ -6,6 +6,7 @@
* https://platform.openai.com/docs/api-reference/chat/create
* Authority (IR): ADR 0003
* Authority (provider dispatch): ADR 0002
* Authority (cache layer): ADR 0005
*
* Design principles (OCP precedent, ESM/.mjs, http built-ins, no external deps):
* - Node ESM, no build step, no bundler
@@ -30,6 +31,8 @@ import {
} from './lib/ir/ir-to-openai.mjs';
import { loadProviders, getProviderForModel, listAllProviderNames } from './lib/providers/index.mjs';
import { ProviderError } from './lib/providers/base.mjs';
import { computeCacheKey, hasCacheControl, extractCacheControlMarkers } from './lib/cache/keys.mjs';
import { CacheStore } from './lib/cache/store.mjs';
// ── Config ────────────────────────────────────────────────────────────────
@@ -45,6 +48,12 @@ const BODY_LIMIT = 5 * 1024 * 1024; // 5 MB
// Empty config → empty loaded map → all POST /v1/chat/completions → 503.
const loadedProviders = loadProviders({ enabled: {} });
// ── Cache layer ───────────────────────────────────────────────────────────
// D1 per-key isolation + D4 singleflight per ADR 0005.
// keyId: '__anonymous__' at D5 — Phase 2 multi-key infrastructure wires in
// the real OLP API key ID here.
export const cacheStore = new CacheStore();
// ── Logging ───────────────────────────────────────────────────────────────
function logEvent(level, event, data = {}) {
@@ -125,21 +134,22 @@ function sendError(res, status, message, type) {
* Returns the standard OLP diagnostic headers.
* Per spec: X-OLP-Provider-Used, X-OLP-Model-Used, X-OLP-Fallback-Hops,
* X-OLP-Cache, X-OLP-Latency-Ms.
* Fallback-Hops is always 0 at D3 (no fallback engine yet — ADR 0004).
* Cache is always 'miss' at D3 (no cache layer yet — ADR 0005).
* Fallback-Hops is always 0 at D5 (no fallback engine yet — ADR 0004).
* Cache reflects actual hit/miss/bypass status from the cache layer (ADR 0005).
*
* @param {object} opts
* @param {string} opts.providerUsed
* @param {string} opts.modelUsed
* @param {number} opts.startMs
* @param {'hit'|'miss'|'bypass'} [opts.cacheStatus='miss']
* @returns {Record<string,string>}
*/
function olpHeaders({ providerUsed, modelUsed, startMs }) {
function olpHeaders({ providerUsed, modelUsed, startMs, cacheStatus = 'miss' }) {
return {
'X-OLP-Provider-Used': providerUsed,
'X-OLP-Model-Used': modelUsed,
'X-OLP-Fallback-Hops': '0',
'X-OLP-Cache': 'miss',
'X-OLP-Cache': cacheStatus,
'X-OLP-Latency-Ms': String(Date.now() - startMs),
};
}
@@ -217,13 +227,88 @@ async function handleChatCompletions(req, res) {
const { provider, name: providerName } = match;
const requestId = generateRequestId();
// Auth context is a stub at D3 — providers will populate this in Phase 1 Day 2+
const authContext = {};
// Auth context is null at D5 — providers fall back to their own credential
// discovery (env var, keychain, credentials file). Phase 2 multi-key
// infrastructure will pass a real authContext carrying the per-key OLP token.
const authContext = null;
const headers = olpHeaders({ providerUsed: providerName, modelUsed: ir.model, startMs });
// ── Cache layer (ADR 0005) ──────────────────────────────────────────────
// keyId: '__anonymous__' at D5. Phase 2 multi-key infrastructure wires the
// real OLP API key ID here for D1 per-key isolation.
const keyId = '__anonymous__';
const cacheKey = computeCacheKey(providerName, ir.model, ir);
// D2 bypass: if the request contains Anthropic cache_control markers,
// skip OLP's response cache entirely (the prompt cache lives at Anthropic's
// side; double-caching would shadow Anthropic's TTLs per ADR 0005 § D2).
//
// Note: hasCacheControl() checks the IR (ADR 0003). Since openAIToIR() does not
// preserve cache_control fields from the raw OpenAI message shape (those fields
// are Anthropic-specific extensions, not part of the IR schema), we also check
// the raw body.messages directly via extractCacheControlMarkers. This ensures
// D2 bypass is triggered even though the IR translator strips the field.
const bypassCache = hasCacheControl(ir) || extractCacheControlMarkers(body?.messages ?? []).length > 0;
if (bypassCache) {
logEvent('debug', 'cache_bypass', { provider: providerName, model: ir.model, reason: 'cache_control_markers' });
}
// ── Collect chunks helper (used by both streaming and non-streaming paths) ──
// collectAllChunks wraps provider.spawn() to collect all IR chunks into an
// array. Used as the computeFn for getOrCompute (D4 singleflight).
//
// Error semantics: if the provider emits a `type: 'error'` chunk, we throw
// a ProviderError instead of returning the chunk array. This prevents
// cache_store.set() from being called on an error-terminated response, per
// ADR 0005 § "Cache write conditions" item 1 ("The response completed
// successfully (no truncation, no error mid-stream)"). The thrown error
// propagates to the catch block at the call site, which returns 502 to
// the client without writing to cache.
async function collectAllChunks() {
const chunks = [];
for await (const irChunk of provider.spawn(ir, authContext)) {
chunks.push(irChunk);
if (irChunk.type === 'error') {
throw new ProviderError(
irChunk.error ?? 'Provider emitted error chunk',
'SPAWN_FAILED',
);
}
if (irChunk.type === 'stop') break;
}
return chunks;
}
// Pre-check: stats-neutral peek before calling getOrCompute so we can
// reliably report hit/miss in the response header. peek() does NOT touch
// hit/miss counters (fix for codex-flagged double-count); getOrCompute()
// increments at the semantic boundary.
const preCheckHit = bypassCache ? false : await cacheStore.peek(keyId, cacheKey);
if (ir.stream) {
// Streaming response path
// D5 simplified D3: cache stores full collected chunks, replays them one
// at a time without timing fidelity. Full D3 (timing-accurate replay) lands
// in a later Phase per ADR 0005 § D3.
let chunks;
let cacheStatus;
try {
if (bypassCache) {
cacheStatus = 'bypass';
chunks = await collectAllChunks();
} else {
// D4 singleflight + D1 per-key isolation
chunks = await cacheStore.getOrCompute(keyId, cacheKey, collectAllChunks);
cacheStatus = preCheckHit ? 'hit' : 'miss';
}
} catch (e) {
logEvent('error', 'spawn_error', { provider: providerName, model: ir.model, error: e.message });
sendError(res, e instanceof ProviderError ? 502 : 500, e.message ?? 'Provider error', 'provider_error');
return;
}
const headers = olpHeaders({ providerUsed: providerName, modelUsed: ir.model, startMs, cacheStatus });
res.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
@@ -232,35 +317,41 @@ async function handleChatCompletions(req, res) {
...headers,
});
try {
for await (const irChunk of provider.spawn(ir, authContext)) {
res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model));
if (irChunk.type === 'stop' || irChunk.type === 'error') break;
}
res.write(SSE_DONE);
} catch (e) {
// Best-effort error reporting in the stream
const errChunk = { type: 'error', error: e.message ?? 'Provider spawn failed' };
res.write(irChunkToOpenAISSE(errChunk, requestId, ir.model));
res.write(SSE_DONE);
logEvent('error', 'spawn_error', { provider: providerName, model: ir.model, error: e.message });
// D3 simplified replay: emit collected chunks sequentially
for (const irChunk of chunks) {
res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model));
if (irChunk.type === 'stop' || irChunk.type === 'error') break;
}
res.write(SSE_DONE);
res.end();
} else {
// Non-streaming response path
let cacheStatus;
let responseObj;
try {
const chunks = [];
for await (const irChunk of provider.spawn(ir, authContext)) {
chunks.push(irChunk);
if (irChunk.type === 'stop' || irChunk.type === 'error') break;
if (bypassCache) {
cacheStatus = 'bypass';
const chunks = await collectAllChunks();
responseObj = irResponseToOpenAINonStream(chunks, requestId, ir.model);
} else {
// D4 singleflight: concurrent identical requests share one spawn.
// We cache the full IR chunk array and assemble the OpenAI response on
// each retrieval so the requestId is fresh per request (per OpenAI spec
// each response has a unique id).
const chunks = await cacheStore.getOrCompute(keyId, cacheKey, collectAllChunks);
cacheStatus = preCheckHit ? 'hit' : 'miss';
responseObj = irResponseToOpenAINonStream(chunks, requestId, ir.model);
}
const response = irResponseToOpenAINonStream(chunks, requestId, ir.model);
sendJSON(res, 200, response, headers);
} catch (e) {
logEvent('error', 'spawn_error', { provider: providerName, model: ir.model, error: e.message });
const status = e instanceof ProviderError ? 502 : 500;
sendError(res, status, e.message ?? 'Provider error', 'provider_error');
return;
}
const headers = olpHeaders({ providerUsed: providerName, modelUsed: ir.model, startMs, cacheStatus });
sendJSON(res, 200, responseObj, headers);
}
}