fix(explore): damp ambient declaration files on flow queries (CG-28)

A file that declares nothing but types and that nothing in the index depends on — a hand-written ambient `.d.ts` of global shims, vendored typings, module augmentation — cannot answer a flow question: no bodies, no call edges, no behaviour, nothing typed by it. But the identifiers it declares are exactly the generic ones a prose question uses (`Body`, `Message`, `ImageMetadata`, `ReadableStream`), so on term overlap it out-scored the implementation. Measured on the new fixture: rank #1 and 51% of delivered source, with the flow's own entry file pushed out of the response entirely. Measured first, per the issue: the Wrangler `worker-configuration.d.ts` that opened this is already handled by CG-25's banner detection, worth 15-46 points of envelope share across four flow queries. CG-25 credited; only the un-bannered case needed anything. `rankPenalty` now multiplies score and graph mass by 0.5 for such files, taken as the STRONGER of it and the generated penalty rather than multiplied — one property two signals see must not be charged twice. Detection is structural, not by extension, and four conditions deep. Two of them were forced by measurement: requiring every symbol to be type-level takes the corpus flag rate from 1-18% (which swept in Kotlin sealed classes, Rust mod.rs re-exports and django's locale tables) down to 0-4%; requiring that nothing depends on the file separates an ambient shim from a working types module, and without it the rule demoted displacement-ts's pipeline `types.ts` and broke the CG-31 gate. A query that NAMES a declared type is exempt, so a question about a type still reaches its declaration at full weight. Precise tokens only, so "…the file body…" cannot exempt a `Body` interface it never meant to name; this needs its own set because `namedSeedIds` is callable-only and a type never becomes one. Regression evidence in docs/benchmarks/explore-declaration-only-cg28.md: 6-repo envelope sweep byte-identical against a clean baseline build, zero ambient files reach the candidate set on VS Code across five queries, corpus flag rate 0-0.74%, both allocation fixtures PASS, full suite 2,978 green. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 14:35:56 -05:00
parent 463f6e7844
commit 9efae0f8f2
18 changed files with 1582 additions and 3 deletions
@@ -0,0 +1,46 @@
+export interface BucketObject {
+  key: string;
+  body: ReadableStream<Uint8Array>;
+  size: number;
+}
+
+export interface Bucket {
+  put(
+    key: string,
+    value: ReadableStream<Uint8Array>,
+    options?: { httpMetadata?: { contentType?: string } },
+  ): Promise<void>;
+  get(key: string): Promise<BucketObject | null>;
+}
+
+export interface MetadataStore {
+  put(id: string, value: string): Promise<void>;
+  get(id: string): Promise<string | null>;
+}
+
+const objects = new Map<string, BucketObject>();
+const rows = new Map<string, string>();
+
+/** The object-storage binding. */
+export function openBucket(): Bucket {
+  return {
+    async put(key, value) {
+      objects.set(key, { key, body: value, size: 0 });
+    },
+    async get(key) {
+      return objects.get(key) ?? null;
+    },
+  };
+}
+
+/** The metadata key-value binding. */
+export function openMetadataStore(): MetadataStore {
+  return {
+    async put(id, value) {
+      rows.set(id, value);
+    },
+    async get(id) {
+      return rows.get(id) ?? null;
+    },
+  };
+}
@@ -0,0 +1,37 @@
+export interface UploadMessageBody {
+  key: string;
+  metadataId: string;
+  contentType: string;
+}
+
+/**
+ * Publish the follow-up message for a stored upload. Batched so a burst of
+ * uploads does not open one producer call per object.
+ */
+export async function enqueueUploadMessage(body: UploadMessageBody): Promise<void> {
+  const queue = openUploadQueue();
+  await queue.send(body, { contentType: 'json' });
+}
+
+/** Consumer side: process a batch of upload messages. */
+export async function consumeUploadBatch(messages: UploadMessageBody[]): Promise<number> {
+  let handled = 0;
+  for (const message of messages) {
+    if (!message.key) continue;
+    handled += 1;
+  }
+  return handled;
+}
+
+interface UploadQueue {
+  send(body: UploadMessageBody, options: { contentType: string }): Promise<void>;
+}
+
+/** The binding lookup, isolated so tests can swap it. */
+export function openUploadQueue(): UploadQueue {
+  return {
+    async send() {
+      /* binding provided by the runtime */
+    },
+  };
+}
@@ -0,0 +1,44 @@
+export interface ParsedUpload {
+  ok: true;
+  key: string;
+  body: ReadableStream<Uint8Array>;
+  contentType: string;
+  width: number;
+  height: number;
+  format: string;
+}
+
+export interface ParseFailure {
+  ok: false;
+  error: string;
+}
+
+/**
+ * Pull the object key, declared dimensions and the raw body stream off an
+ * upload request. Never buffers the body — the stream is handed straight to
+ * the storage layer.
+ */
+export async function parseUploadRequest(
+  request: Request,
+): Promise<ParsedUpload | ParseFailure> {
+  const url = new URL(request.url);
+  const key = url.searchParams.get('key');
+  if (!key) return { ok: false, error: 'missing key' };
+  if (!request.body) return { ok: false, error: 'missing body' };
+
+  return {
+    ok: true,
+    key,
+    body: request.body as ReadableStream<Uint8Array>,
+    contentType: request.headers.get('content-type') ?? 'application/octet-stream',
+    width: numberParam(url, 'width'),
+    height: numberParam(url, 'height'),
+    format: url.searchParams.get('format') ?? 'jpeg',
+  };
+}
+
+function numberParam(url: URL, name: string): number {
+  const raw = url.searchParams.get(name);
+  const parsed = raw ? Number.parseInt(raw, 10) : 0;
+  return Number.isFinite(parsed) ? parsed : 0;
+}
@@ -0,0 +1,66 @@
+import { streamBodyToStorage } from '../storage/stream.js';
+import { recordImageMetadata } from '../storage/metadata.js';
+import { enqueueUploadMessage } from '../lib/queue.js';
+import { parseUploadRequest } from '../lib/request.js';
+
+export interface UploadResult {
+  key: string;
+  bytes: number;
+  contentType: string;
+}
+
+/**
+ * Entry point for an upload request: parse it, stream the body into object
+ * storage, record the image metadata, then queue the follow-up work.
+ */
+export async function handleUploadRequest(request: Request): Promise<Response> {
+  const parsed = await parseUploadRequest(request);
+  if (!parsed.ok) {
+    return new Response(JSON.stringify({ error: parsed.error }), { status: 400 });
+  }
+
+  const stored = await streamBodyToStorage(parsed.body, parsed.key, parsed.contentType);
+  const metadata = await recordImageMetadata(stored.key, {
+    width: parsed.width,
+    height: parsed.height,
+    format: parsed.format,
+    bytes: stored.bytes,
+  });
+
+  await enqueueUploadMessage({
+    key: stored.key,
+    metadataId: metadata.id,
+    contentType: stored.contentType,
+  });
+
+  return new Response(JSON.stringify(summarizeUpload(stored, metadata.id)), {
+    status: 201,
+    headers: { 'content-type': 'application/json' },
+  });
+}
+
+/** Shape the client sees back after a successful upload. */
+export function summarizeUpload(stored: UploadResult, metadataId: string) {
+  return {
+    key: stored.key,
+    bytes: stored.bytes,
+    contentType: stored.contentType,
+    metadataId,
+  };
+}
+
+/** Reject uploads whose declared size exceeds the per-account ceiling. */
+export function isWithinUploadLimit(bytes: number, limit: number): boolean {
+  if (!Number.isFinite(bytes) || bytes < 0) return false;
+  return bytes <= limit;
+}
+
+/** Delete-side counterpart, kept here so the route module is not a one-liner. */
+export async function handleDeleteRequest(request: Request, key: string): Promise<Response> {
+  const parsed = await parseUploadRequest(request);
+  if (!parsed.ok) {
+    return new Response(JSON.stringify({ error: parsed.error }), { status: 400 });
+  }
+  await enqueueUploadMessage({ key, metadataId: '', contentType: 'application/x-delete' });
+  return new Response(null, { status: 204 });
+}
@@ -0,0 +1,54 @@
+import { openMetadataStore } from '../lib/bucket.js';
+
+export interface ImageMetadataInput {
+  width: number;
+  height: number;
+  format: string;
+  bytes: number;
+}
+
+export interface ImageMetadataRecord extends ImageMetadataInput {
+  id: string;
+  key: string;
+  recordedAt: number;
+}
+
+/**
+ * Record the image metadata for a stored object. Writes go to the metadata
+ * store keyed by object key; the returned record carries the id the queue
+ * message references.
+ */
+export async function recordImageMetadata(
+  key: string,
+  input: ImageMetadataInput,
+): Promise<ImageMetadataRecord> {
+  const store = openMetadataStore();
+  const record: ImageMetadataRecord = {
+    ...input,
+    id: metadataIdFor(key, input),
+    key,
+    recordedAt: 0,
+  };
+  await store.put(record.id, JSON.stringify(record));
+  return record;
+}
+
+/** Deterministic id so a retried upload records the same metadata row. */
+export function metadataIdFor(key: string, input: ImageMetadataInput): string {
+  return `${key}:${input.format}:${input.width}x${input.height}`;
+}
+
+/** Read a metadata record back for the download and listing paths. */
+export async function loadImageMetadata(id: string): Promise<ImageMetadataRecord | null> {
+  const store = openMetadataStore();
+  const raw = await store.get(id);
+  return raw ? (JSON.parse(raw) as ImageMetadataRecord) : null;
+}
+
+/** Normalize a client-declared format string to the canonical set. */
+export function normalizeFormat(format: string): string {
+  const lowered = format.trim().toLowerCase();
+  if (lowered === 'jpg') return 'jpeg';
+  if (lowered === 'tif') return 'tiff';
+  return lowered;
+}
@@ -0,0 +1,86 @@
+import { openBucket } from '../lib/bucket.js';
+import type { StorageFailure, UploadTelemetry } from './types.js';
+
+export interface StoredObject {
+  key: string;
+  bytes: number;
+  contentType: string;
+}
+
+/**
+ * Stream a request body into object storage without buffering it in memory.
+ * The body is piped through a counting transform so the byte total is known
+ * by the time the put resolves.
+ */
+export async function streamBodyToStorage(
+  body: ReadableStream<Uint8Array>,
+  key: string,
+  contentType: string,
+): Promise<StoredObject> {
+  const bucket = openBucket();
+  const counter = createByteCounter();
+  const piped = body.pipeThrough(counter.transform, { preventClose: false });
+
+  await bucket.put(key, piped, { httpMetadata: { contentType } });
+
+  return { key, bytes: counter.total(), contentType };
+}
+
+/**
+ * A transform stream that counts the bytes flowing through it. Separated from
+ * the pipe above so the byte total can be read after the stream settles.
+ */
+export function createByteCounter() {
+  let total = 0;
+  const transform = new TransformStream<Uint8Array, Uint8Array>({
+    transform(chunk, controller) {
+      total += chunk.byteLength;
+      controller.enqueue(chunk);
+    },
+  });
+  return { transform, total: () => total };
+}
+
+/**
+ * Read a stored object back out of the bucket as a stream, for the download
+ * path. Mirrors the upload side so both directions live in one module.
+ */
+export async function readObjectStream(key: string): Promise<ReadableStream<Uint8Array> | null> {
+  const bucket = openBucket();
+  const object = await bucket.get(key);
+  if (!object) return null;
+  return object.body;
+}
+
+/** Timing/retry record for one stored object, handed to the metrics sink. */
+export function telemetryFor(stored: StoredObject, durationMs: number): UploadTelemetry {
+  return { key: stored.key, bytes: stored.bytes, durationMs, retries: 0 };
+}
+
+/** Describe a failed stage so the caller can report it without re-deriving it. */
+export function storageFailure(
+  key: string,
+  stage: StorageFailure['stage'],
+  message: string,
+): StorageFailure {
+  return { key, stage, message };
+}
+
+/** Cap a stream at `limit` bytes, erroring out rather than storing an overrun. */
+export function limitStream(
+  source: ReadableStream<Uint8Array>,
+  limit: number,
+): ReadableStream<Uint8Array> {
+  let seen = 0;
+  const guard = new TransformStream<Uint8Array, Uint8Array>({
+    transform(chunk, controller) {
+      seen += chunk.byteLength;
+      if (seen > limit) {
+        controller.error(new Error(`upload exceeded ${limit} bytes`));
+        return;
+      }
+      controller.enqueue(chunk);
+    },
+  });
+  return source.pipeThrough(guard);
+}
@@ -0,0 +1,18 @@
+/**
+ * Shared shapes for the storage layer. Declaration-only like the ambient files
+ * under `types/` — but the modules that answer a flow question are typed BY it,
+ * so it is part of that answer's structure rather than a global shim.
+ */
+
+export interface UploadTelemetry {
+  key: string;
+  bytes: number;
+  durationMs: number;
+  retries: number;
+}
+
+export interface StorageFailure {
+  key: string;
+  stage: 'parse' | 'stream' | 'metadata' | 'queue';
+  message: string;
+}