Cache or Recompute? The Agentic Pipeline Decision That Determines Whether Your Real-Time Agent Feels Fast or Broken

// TypeScript: Structured cache retrieval with provenance trace emission
import { tracer } from './observability';

interface CachedResult<T> {
  value: T;
  computedAt: number;
  cacheKey: string;
  provenance: { agentId: string; modelId: string; inputHash: string };
}

async function getWithProvenance<T>(
  key: string,
  recompute: () => Promise<CachedResult<T>>,
  maxAgeMs: number
): Promise<T> {
  const span = tracer.startSpan('cache.lookup', { attributes: { 'cache.key': key } });
  const cached = await cacheStore.get<CachedResult<T>>(key);

  if (cached) {
    const ageMs = Date.now() - cached.computedAt;
    span.setAttributes({ 'cache.hit': true, 'cache.age_ms': ageMs });

    if (ageMs > maxAgeMs) {
      span.addEvent('cache.stale_escalation', { 'cache.key': key, 'cache.age_ms': ageMs });
      span.end();
      // Route to human review gate rather than silently serving stale data
      await humanReviewQueue.push({ key, cached, reason: 'stale_cache_action' });
    }

    span.end();
    return cached.value;
  }

  span.setAttributes({ 'cache.hit': false });
  const fresh = await recompute();
  await cacheStore.set(key, fresh);
  span.end();
  return fresh.value;
}

// TypeScript: Semantic diff fingerprinting for review verdict caching
interface DiffFingerprint {
  cyclomaticDelta: number;
  securitySurface: string[]; // e.g. ['sql_concat', 'user_input_unsanitized']
  dependencyDelta: string[];
  testCoverageRatio: number;
}

interface VerdictCacheEntry {
  fingerprint: DiffFingerprint;
  verdict: 'approved' | 'rejected' | 'needs_changes';
  reviewedBy: string;
  reviewedAt: number;
  notes: string;
}

function fingerprintSimilarity(a: DiffFingerprint, b: DiffFingerprint): number {
  const securityMatch = a.securitySurface.every(s => b.securitySurface.includes(s)) ? 1 : 0;
  const complexityProximity = 1 - Math.min(Math.abs(a.cyclomaticDelta - b.cyclomaticDelta) / 10, 1);
  const coverageProximity = 1 - Math.abs(a.testCoverageRatio - b.testCoverageRatio);
  return (securityMatch * 0.5) + (complexityProximity * 0.3) + (coverageProximity * 0.2);
}

async function lookupVerdictCache(
  fingerprint: DiffFingerprint,
  threshold = 0.88
): Promise<VerdictCacheEntry | null> {
  const candidates = await verdictStore.getRecent(50);
  const best = candidates
    .map(entry => ({ entry, score: fingerprintSimilarity(fingerprint, entry.fingerprint) }))
    .filter(({ score }) => score >= threshold)
    .sort((a, b) => b.score - a.score)[0];
  return best?.entry ?? null;
}

// TypeScript: Stateless Lambda agent with layered cache telemetry
import { BedrockRuntimeClient, InvokeModelCommand } from '@aws-sdk/client-bedrock-runtime';
import { DynamoDBDocumentClient, GetCommand, PutCommand } from '@aws-sdk/lib-dynamodb';

const bedrock = new BedrockRuntimeClient({ region: process.env.AWS_REGION });
const dynamo = DynamoDBDocumentClient.from(/* client */);

export async function handleAgentTurn(event: { sessionId: string; userMessage: string }) {
  const start = Date.now();

  // Layer 1: Conversation state retrieval
  const historyResult = await dynamo.send(new GetCommand({
    TableName: 'agent-sessions',
    Key: { sessionId: event.sessionId },
  }));
  const historyMs = Date.now() - start;

  const history = historyResult.Item?.turns ?? [];
  const prompt = buildPrompt(history, event.userMessage);

  // Layer 2: Model inference (Bedrock tracks its own prompt cache internally)
  const inferenceStart = Date.now();
  const response = await bedrock.send(new InvokeModelCommand({
    modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    body: JSON.stringify({ messages: prompt, max_tokens: 1024 }),
    contentType: 'application/json',
  }));
  const inferenceMs = Date.now() - inferenceStart;

  // Emit structured latency breakdown — not a single 'duration' metric
  console.log(JSON.stringify({
    sessionId: event.sessionId,
    latency: { historyFetchMs: historyMs, inferenceMs, totalMs: Date.now() - start },
    cacheSignals: { promptTokenCount: response.$metadata.requestId },
  }));

  const agentReply = parseResponse(response.body);

  await dynamo.send(new PutCommand({
    TableName: 'agent-sessions',
    Item: { sessionId: event.sessionId, turns: [...history, { user: event.userMessage, agent: agentReply }] },
  }));

  return agentReply;
}

// TypeScript: Per-layer cache telemetry for GenAI platform pipeline
type LayerName = 'guardrail' | 'retrieval' | 'routing' | 'inference' | 'output';

interface LayerTelemetry {
  layer: LayerName;
  cacheHit: boolean;
  latencyMs: number;
  cacheKey?: string;
  cacheAgeMs?: number;
}

class PipelineTelemetryCollector {
  private events: LayerTelemetry[] = [];

  record(event: LayerTelemetry) {
    this.events.push(event);
  }

  flush(requestId: string) {
    const totalMs = this.events.reduce((sum, e) => sum + e.latencyMs, 0);
    const cacheHitRate = this.events.filter(e => e.cacheHit).length / this.events.length;

    console.log(JSON.stringify({
      requestId,
      pipeline_latency_breakdown: this.events.map(e => ({
        layer: e.layer,
        cache_hit: e.cacheHit,
        latency_ms: e.latencyMs,
        cache_age_ms: e.cacheAgeMs ?? null,
      })),
      total_latency_ms: totalMs,
      cache_hit_rate: cacheHitRate,
      // Flag requests where inference was NOT the dominant latency
      inference_bottleneck: this.events.find(e => e.layer === 'inference')!.latencyMs ===
        Math.max(...this.events.map(e => e.latencyMs)),
    }));
  }
}

// TypeScript: Token spend anomaly detection with cache hit rate correlation
interface TokenSpendEvent {
  agentId: string;
  modelProvider: string;
  promptTokens: number;
  completionTokens: number;
  cacheHit: boolean;
  requestId: string;
  timestamp: number;
}

class SpendAnomalyDetector {
  private windowMs = 5 * 60 * 1000; // 5-minute rolling window
  private expectedCacheHitRate = 0.65; // baseline from historical data

  async evaluate(recent: TokenSpendEvent[]): Promise<void> {
    const windowStart = Date.now() - this.windowMs;
    const inWindow = recent.filter(e => e.timestamp >= windowStart);

    const cacheHitRate = inWindow.filter(e => e.cacheHit).length / inWindow.length;
    const totalSpend = inWindow.reduce((sum, e) => sum + e.promptTokens + e.completionTokens, 0);

    // A sudden drop in cache hit rate without a code deployment is suspicious
    if (cacheHitRate < this.expectedCacheHitRate * 0.7) {
      await this.escalate({
        signal: 'cache_bypass_anomaly',
        observedHitRate: cacheHitRate,
        expectedHitRate: this.expectedCacheHitRate,
        totalTokensInWindow: totalSpend,
        possibleCauses: ['relay_proxy_stripping_cache_headers', 'cache_key_config_regression', 'new_agent_traffic_pattern'],
      });
    }
  }

  private async escalate(payload: object) {
    // Route to governance event bus, not just a log line
    await governanceEventBus.publish('spend.anomaly.detected', payload);
  }
}