[codex] Enable trusted trace object upload authority (#3923)

* Enable trusted trace object upload authority

* Fix trace object authority registration edge cases

Generated-By: looper 0.9.4 (runner=fixer, agent=codex)

apps/daemon/src/langfuse-bridge.ts

@@ -32,6 +32,7 @@ import {
   type MessageSummary,
   type ReportContext,
   type RuntimeInfo,
+  type TelemetrySinkConfig,
   type ToolCallSummary,
   type TurnInfo,
 } from './langfuse-trace.js';
@@ -168,6 +169,46 @@ function mergeTraceSafeManifests(
   };
 }
 
+function inferObjectRegistrationRelayUrl(env: NodeJS.ProcessEnv = process.env): string | null {
+  const objectRelayUrl = env.OPEN_DESIGN_OBJECT_RELAY_URL?.trim();
+  if (!objectRelayUrl) return null;
+  try {
+    const url = new URL(objectRelayUrl);
+    url.pathname = url.pathname.replace(/\/api\/objects\/batch\/?$/, '/api/langfuse');
+    return url.toString().replace(/\/+$/, '');
+  } catch {
+    return objectRelayUrl.replace(/\/api\/objects\/batch\/?$/, '/api/langfuse').replace(/\/+$/, '');
+  }
+}
+
+function parsePositiveInt(value: string | undefined, fallback: number): number {
+  if (value === undefined) return fallback;
+  const parsed = Number.parseInt(value, 10);
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+}
+
+function parseNonNegativeInt(value: string | undefined, fallback: number): number {
+  if (value === undefined) return fallback;
+  const parsed = Number.parseInt(value, 10);
+  return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback;
+}
+
+function objectRegistrationTelemetryConfig(
+  env: NodeJS.ProcessEnv = process.env,
+): Extract<TelemetrySinkConfig, { kind: 'relay' }> | null {
+  const relayUrl = inferObjectRegistrationRelayUrl(env);
+  if (!relayUrl) return null;
+  return {
+    kind: 'relay',
+    relayUrl,
+    timeoutMs: parsePositiveInt(
+      env.OPEN_DESIGN_OBJECT_RELAY_TIMEOUT_MS ?? env.OPEN_DESIGN_TELEMETRY_TIMEOUT_MS,
+      20_000,
+    ),
+    retries: parseNonNegativeInt(env.OPEN_DESIGN_TELEMETRY_RETRIES, 1),
+  };
+}
+
 function turnInfoFromRun(
   run: DaemonRunRecord,
   agentReportedModel: string | null,
@@ -874,7 +915,7 @@ export async function reportRunCompletedFromDaemon(
       attachmentsRaw,
       producedFilesRaw,
     });
-    const uploadedManifests = await buildTraceObjectManifests({
+    const objectManifestOptions = {
       installationId,
       projectId: run.projectId ?? '',
       runId: run.id,
@@ -885,9 +926,8 @@ export async function reportRunCompletedFromDaemon(
       prompt: telemetryPrompt,
       prefs,
       ...(opts.fetchImpl ? { fetchImpl: opts.fetchImpl } : {}),
-    });
-    const finalManifests = mergeTraceSafeManifests(manifests, uploadedManifests);
-    const ctx: ReportContext = {
+    } satisfies Parameters<typeof buildTraceObjectManifests>[0];
+    const buildContext = (finalManifests: FinalTraceSafeManifests): ReportContext => ({
       installationId,
       projectId: run.projectId ?? '',
       conversationId: run.conversationId ?? '',
@@ -928,10 +968,26 @@ export async function reportRunCompletedFromDaemon(
       ...(turn ? { turn } : {}),
       runtime,
       ...(run.promptTelemetry ? { promptTelemetry: run.promptTelemetry } : {}),
-    };
+    });
 
+    const registrationManifests = await buildTraceObjectManifests({
+      ...objectManifestOptions,
+      uploadMode: 'manifest-only',
+    });
+    if (registrationManifests) {
+      await reportRunCompleted(
+        buildContext(mergeTraceSafeManifests(manifests, registrationManifests)),
+        {
+          config: objectRegistrationTelemetryConfig(),
+          ...(opts.fetchImpl ? { fetchImpl: opts.fetchImpl } : {}),
+        },
+      );
+    }
+
+    const uploadedManifests = await buildTraceObjectManifests(objectManifestOptions);
+    const finalManifests = mergeTraceSafeManifests(manifests, uploadedManifests);
     await reportRunCompleted(
-      ctx,
+      buildContext(finalManifests),
       opts.fetchImpl ? { fetchImpl: opts.fetchImpl } : {},
     );
   } catch (err) {

apps/daemon/src/trace-object-manifest.ts

@@ -61,6 +61,7 @@ export interface BuildTraceObjectManifestsOptions {
   fetchImpl?: typeof fetch;
   env?: NodeJS.ProcessEnv;
   now?: () => Date;
+  uploadMode?: 'manifest-only' | 'upload';
 }
 
 export interface TraceArtifactObjectSource {
@@ -136,15 +137,7 @@ function storageRef(projectId: string, runId: string, objectClass: ObjectClass,
 function inferRelayUrl(env: NodeJS.ProcessEnv): string | null {
   const explicit = env.OPEN_DESIGN_OBJECT_RELAY_URL?.trim();
   if (explicit) return explicit.replace(/\/+$/, '');
-  const telemetry = env.OPEN_DESIGN_TELEMETRY_RELAY_URL?.trim();
-  if (!telemetry) return null;
-  try {
-    const url = new URL(telemetry);
-    url.pathname = url.pathname.replace(/\/api\/langfuse\/?$/, '/api/objects/batch');
-    return url.toString().replace(/\/+$/, '');
-  } catch {
-    return null;
-  }
+  return null;
 }
 
 function inferAuthorizeUrl(batchUrl: string): string {
@@ -169,7 +162,7 @@ function readRelayConfig(env: NodeJS.ProcessEnv): ObjectRelayConfig | null {
   return {
     url,
     authorizeUrl: inferAuthorizeUrl(url),
-    uploadsEnabled: env.NODE_ENV === 'test',
+    uploadsEnabled: true,
     timeoutMs: parsePositiveInt(
       env.OPEN_DESIGN_OBJECT_RELAY_TIMEOUT_MS ?? env.OPEN_DESIGN_TELEMETRY_TIMEOUT_MS,
       10_000,
@@ -250,6 +243,18 @@ function manifestBase(
   };
 }
 
+function mergeSourceDigest(
+  entry: TraceObjectManifestEntry,
+  source: TraceObjectSource,
+): TraceObjectManifestEntry {
+  if (!source.body) return entry;
+  return {
+    ...entry,
+    size_bytes: source.body.byteLength,
+    sha256: sha256(source.body),
+  };
+}
+
 function buildObjectBatchBody(
   opts: BuildTraceObjectManifestsOptions,
   objects: ObjectRelayRequestObject[],
@@ -640,6 +645,14 @@ export async function buildTraceObjectManifests(
   if (sources.length === 0) return undefined;
 
   const manifests = sources.map((source) => manifestBase(source, opts, now));
+  if (opts.uploadMode === 'manifest-only') {
+    return groupManifests(manifests.map((entry, index) => ({
+      ...mergeSourceDigest(entry, sources[index]!),
+      status: 'unavailable' as const,
+      stored_in_open_design: false,
+      reason: entry.reason ?? 'relay_authorization_pending',
+    })));
+  }
 
   const relayResults = await postObjectBatch(config, opts, manifests, sources);
   const resultByRef = new Map(relayResults.map((result) => [result.storage_ref, result]));

apps/daemon/tests/langfuse-bridge.test.ts

@@ -578,6 +578,7 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
     });
 
     process.env.OPEN_DESIGN_OBJECT_RELAY_URL = 'https://telemetry.open-design.ai/api/objects/batch';
+    process.env.OPEN_DESIGN_TELEMETRY_RELAY_URL = 'https://telemetry.open-design.ai/api/langfuse';
     process.env.LANGFUSE_PUBLIC_KEY = 'pk';
     process.env.LANGFUSE_SECRET_KEY = 'sk';
     try {
@@ -613,14 +614,17 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
       });
     } finally {
       delete process.env.OPEN_DESIGN_OBJECT_RELAY_URL;
+      delete process.env.OPEN_DESIGN_TELEMETRY_RELAY_URL;
       delete process.env.LANGFUSE_PUBLIC_KEY;
       delete process.env.LANGFUSE_SECRET_KEY;
     }
 
-    expect(fetchSpy).toHaveBeenCalledTimes(3);
-    expect(fetchSpy.mock.calls[0]![0]).toContain('/api/objects/authorize');
-    expect(fetchSpy.mock.calls[1]![0]).toContain('/api/objects/batch');
-    const langfuseInit = fetchSpy.mock.calls[2]![1] as RequestInit;
+    expect(fetchSpy).toHaveBeenCalledTimes(4);
+    expect(fetchSpy.mock.calls[0]![0]).toContain('/api/langfuse');
+    expect(fetchSpy.mock.calls[1]![0]).toContain('/api/objects/authorize');
+    expect(fetchSpy.mock.calls[2]![0]).toContain('/api/objects/batch');
+    expect(fetchSpy.mock.calls[3]![0]).toContain('/api/langfuse');
+    const langfuseInit = fetchSpy.mock.calls[3]![1] as RequestInit;
     const langfuseBody = langfuseInit.body as string;
     expect(langfuseBody).not.toContain('attachment body should stay out of langfuse');
     expect(langfuseBody).not.toContain('<!doctype html><h1>artifact body</h1>');
@@ -658,7 +662,7 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
     );
   });
 
-  it('derives manifest completeness from merged uploaded and fallback manifests', async () => {
+  it('registers object upload authority through the object relay when traces use direct Langfuse', async () => {
     await writeAppCfg({
       installationId: 'install-uuid-1',
       telemetry: { metrics: true, content: true, artifactManifest: true },
@@ -697,6 +701,89 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
     process.env.OPEN_DESIGN_OBJECT_RELAY_URL = 'https://telemetry.open-design.ai/api/objects/batch';
     process.env.LANGFUSE_PUBLIC_KEY = 'pk';
     process.env.LANGFUSE_SECRET_KEY = 'sk';
+    try {
+      await reportRunCompletedFromDaemon({
+        db: makeDbWithListMessages({
+          'conv-1': [
+            { id: 'user-1', role: 'user', content: 'Build it.' },
+            {
+              id: 'msg-1',
+              role: 'assistant',
+              content: 'done',
+              producedFiles: [{ name: 'index.html', kind: 'html', size: 35 }],
+            },
+          ],
+        }),
+        dataDir,
+        run: makeRun() as any,
+        fetchImpl: fetchSpy as any,
+      });
+    } finally {
+      delete process.env.OPEN_DESIGN_OBJECT_RELAY_URL;
+      delete process.env.LANGFUSE_PUBLIC_KEY;
+      delete process.env.LANGFUSE_SECRET_KEY;
+    }
+
+    expect(fetchSpy).toHaveBeenCalledTimes(4);
+    expect(fetchSpy.mock.calls[0]![0]).toBe('https://telemetry.open-design.ai/api/langfuse');
+    expect(fetchSpy.mock.calls[1]![0]).toBe('https://telemetry.open-design.ai/api/objects/authorize');
+    expect(fetchSpy.mock.calls[2]![0]).toBe('https://telemetry.open-design.ai/api/objects/batch');
+    expect(fetchSpy.mock.calls[3]![0]).toBe('https://us.cloud.langfuse.com/api/public/ingestion');
+    const registrationBatch = JSON.parse(fetchSpy.mock.calls[0]![1]!.body as string).batch as any[];
+    const finalBatch = JSON.parse(fetchSpy.mock.calls[3]![1]!.body as string).batch as any[];
+    expect(registrationBatch[0].body.metadata.artifact_manifest[0]).toMatchObject({
+      object_class: 'artifact',
+      storage_ref: expect.stringContaining(
+        'od://objects/workspaces/unknown/projects/proj-1/runs/run-id-1/artifact/',
+      ),
+    });
+    expect(finalBatch[0].body.metadata.artifact_manifest[0]).toMatchObject({
+      object_class: 'artifact',
+      status: 'ok',
+      stored_in_open_design: true,
+    });
+  });
+
+  it('derives manifest completeness from merged uploaded and fallback manifests', async () => {
+    await writeAppCfg({
+      installationId: 'install-uuid-1',
+      telemetry: { metrics: true, content: true, artifactManifest: true },
+    });
+    const projectDir = path.join(dataDir, 'projects', 'proj-1');
+    await mkdir(projectDir, { recursive: true });
+    await writeFile(path.join(projectDir, 'index.html'), '<!doctype html><h1>artifact body</h1>');
+    const fetchSpy = vi.fn(async (url: string, init: RequestInit) => {
+      if (url.includes('/api/objects/authorize')) {
+        const parsed = JSON.parse(init.body as string) as {
+          objects: Array<{ storage_ref: string; object_class: string }>;
+        };
+        expect(parsed.objects).toHaveLength(1);
+        expect(parsed.objects[0]).toMatchObject({ object_class: 'artifact' });
+        return new Response(JSON.stringify({ upload_token: 'upload-token' }), { status: 200 });
+      }
+      if (url.includes('/api/objects/batch')) {
+        const parsed = JSON.parse(init.body as string) as {
+          objects: Array<{ storage_ref: string; content_base64: string }>;
+        };
+        return new Response(
+          JSON.stringify({
+            objects: parsed.objects.map((object) => ({
+              storage_ref: object.storage_ref,
+              status: 'available',
+              size_bytes: Buffer.from(object.content_base64, 'base64').byteLength,
+              sha256: 'sha256:uploaded-artifact',
+            })),
+          }),
+          { status: 200 },
+        );
+      }
+      return new Response('{}', { status: 207 });
+    });
+
+    process.env.OPEN_DESIGN_OBJECT_RELAY_URL = 'https://telemetry.open-design.ai/api/objects/batch';
+    process.env.OPEN_DESIGN_TELEMETRY_RELAY_URL = 'https://telemetry.open-design.ai/api/langfuse';
+    process.env.LANGFUSE_PUBLIC_KEY = 'pk';
+    process.env.LANGFUSE_SECRET_KEY = 'sk';
     try {
       await reportRunCompletedFromDaemon({
         db: makeDbWithListMessages({
@@ -721,12 +808,16 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
       });
     } finally {
       delete process.env.OPEN_DESIGN_OBJECT_RELAY_URL;
+      delete process.env.OPEN_DESIGN_TELEMETRY_RELAY_URL;
       delete process.env.LANGFUSE_PUBLIC_KEY;
       delete process.env.LANGFUSE_SECRET_KEY;
     }
 
-    expect(fetchSpy).toHaveBeenCalledTimes(3);
-    const langfuseInit = fetchSpy.mock.calls[2]![1] as RequestInit;
+    expect(fetchSpy).toHaveBeenCalledTimes(4);
+    expect(fetchSpy.mock.calls[0]![0]).toContain('/api/langfuse');
+    expect(fetchSpy.mock.calls[1]![0]).toContain('/api/objects/authorize');
+    expect(fetchSpy.mock.calls[2]![0]).toContain('/api/objects/batch');
+    const langfuseInit = fetchSpy.mock.calls[3]![1] as RequestInit;
     const batch = JSON.parse(langfuseInit.body as string).batch as any[];
     const trace = batch[0].body;
     expect(trace.metadata.manifest_completeness).toBe('partial');

apps/telemetry-worker/README.md

@@ -18,11 +18,12 @@ them through the `TRACE_OBJECT_BUCKET` R2 binding, and returns trace-safe
 `storage_ref` / `sha256` / size metadata for Langfuse manifests.
 
 Object ingest accepts a short-lived upload token signed by Worker-held
-authority. Public metadata-only authorization is disabled until the Worker can
-verify a trusted telemetry authority for the requested objects. The long-lived
-signing secret stays in the Worker and is never packaged into the daemon/client.
-Until that trusted authority exists, production daemon telemetry emits trace-safe
-object manifests only and does not attempt object authorization or upload.
+authority. The Worker issues that token only after the same Worker has already
+seen the run's trace-safe object scope in a normal `/api/langfuse` telemetry
+batch and stored it in `TRACE_OBJECT_SCOPE_KV`. Caller-supplied metadata plus
+the public marker header is not enough to authorize a production upload. The
+long-lived signing secret stays in the Worker and is never packaged into the
+daemon/client.
 
 Local development can bypass the relay by setting direct `LANGFUSE_PUBLIC_KEY`
 and `LANGFUSE_SECRET_KEY` environment variables for the daemon. Packaged
@@ -42,13 +43,12 @@ Rate Limiting bindings for two independent keys:
 Object ingest uses the same rate limit bindings with a separate marker value,
 `X-Open-Design-Telemetry: object-ingestion-v1`. `POST /api/objects/batch` must
 include a signed upload token, and the Worker re-checks the namespace, size, and
-sha256 before writing to R2. `POST /api/objects/authorize` does not issue
-production upload tokens from caller-supplied metadata; it is limited to the
-explicit `TRACE_OBJECT_AUTHORIZE_TEST_ONLY=1` harness until a server-verifiable
-telemetry authority exists. The Worker also applies IP rate limiting before
-reading object bodies. It enforces a 10 MiB single-object limit and a 20 MiB
-request-body limit by default. Oversized objects are reported as unavailable
-instead of being written.
+sha256 before writing to R2. `POST /api/objects/authorize` reads only bounded
+metadata, checks that every requested object exactly matches a previously
+registered telemetry scope, then returns a five-minute token. The Worker also
+applies IP rate limiting before reading object bodies. It enforces a 10 MiB
+single-object limit and a 20 MiB request-body limit by default. Oversized
+objects are reported as unavailable instead of being written.
 
 ## Secrets
 
@@ -69,6 +69,10 @@ packaged client or daemon. Required worker configuration:
 binding = "TRACE_OBJECT_BUCKET"
 bucket_name = "open-design-observability"
 
+[[kv_namespaces]]
+binding = "TRACE_OBJECT_SCOPE_KV"
+id = "<cloudflare-kv-namespace-id>"
+
 [vars]
 TRACE_OBJECT_PREFIX = "observability"
 TRACE_OBJECT_MAX_BYTES = "10485760"

apps/telemetry-worker/src/index.ts

@@ -8,6 +8,7 @@ const RELAY_MARKER_VALUE = 'langfuse-ingestion-v1';
 const OBJECT_RELAY_MARKER_VALUE = 'object-ingestion-v1';
 const OBJECT_UPLOAD_TOKEN_TTL_SECONDS = 5 * 60;
 const OBJECT_AUTHORIZE_MAX_OBJECTS = 1000;
+const OBJECT_SCOPE_TTL_SECONDS = 10 * 60;
 const ALLOWED_EVENT_TYPES = new Set([
   'trace-create',
   'span-create',
@@ -33,6 +34,15 @@ interface R2BucketBinding {
   ): Promise<unknown>;
 }
 
+interface ObjectScopeBinding {
+  get(key: string): Promise<string | null>;
+  put(
+    key: string,
+    value: string,
+    options?: { expirationTtl?: number },
+  ): Promise<unknown>;
+}
+
 export interface Env {
   LANGFUSE_PUBLIC_KEY?: string;
   LANGFUSE_SECRET_KEY?: string;
@@ -42,7 +52,7 @@ export interface Env {
   TRACE_OBJECT_MAX_BYTES?: string;
   TRACE_OBJECT_BATCH_MAX_BYTES?: string;
   TRACE_OBJECT_UPLOAD_SECRET?: string;
-  TRACE_OBJECT_AUTHORIZE_TEST_ONLY?: string;
+  TRACE_OBJECT_SCOPE_KV?: ObjectScopeBinding;
   TELEMETRY_CLIENT_RATE_LIMITER?: RateLimitBinding;
   TELEMETRY_IP_RATE_LIMITER?: RateLimitBinding;
 }
@@ -142,6 +152,11 @@ function objectScopeKey(object: Pick<ObjectUploadScopeObject, 'storage_ref' | 'o
   return `${object.object_class}\n${object.storage_ref}`;
 }
 
+function objectAuthorityRegistryKey(clientId: string, projectId: string, runId: string): string {
+  const raw = JSON.stringify({ clientId, projectId, runId });
+  return `trace-object-scope:${base64UrlEncode(raw)}`;
+}
+
 function validateObjectScopeBody(value: unknown, maxObjects: number): string | null {
   if (!isRecord(value)) return 'body must be a JSON object';
   if (typeof value.client_id !== 'string' || value.client_id.length === 0) {
@@ -364,8 +379,8 @@ function hasObjectRelayConfig(env: Env): boolean {
   return Boolean(env.TRACE_OBJECT_BUCKET && env.TRACE_OBJECT_UPLOAD_SECRET?.trim());
 }
 
-function allowTestObjectAuthorize(env: Env): boolean {
-  return env.TRACE_OBJECT_AUTHORIZE_TEST_ONLY === '1';
+function hasObjectAuthorizeConfig(env: Env): boolean {
+  return hasObjectRelayConfig(env) && Boolean(env.TRACE_OBJECT_SCOPE_KV);
 }
 
 function isHealthPath(request: Request): boolean {
@@ -426,6 +441,56 @@ function decodeBase64(input: string): Uint8Array | null {
   }
 }
 
+function traceCreateEventIds(value: unknown): Set<string> {
+  const ids = new Set<string>();
+  if (!isRecord(value) || !Array.isArray(value.batch)) return ids;
+  for (const event of value.batch) {
+    if (
+      isRecord(event) &&
+      event.type === 'trace-create' &&
+      typeof event.id === 'string' &&
+      event.id.length > 0
+    ) {
+      ids.add(event.id);
+    }
+  }
+  return ids;
+}
+
+function eventResultIds(value: unknown): Set<string> {
+  const ids = new Set<string>();
+  if (!Array.isArray(value)) return ids;
+  for (const item of value) {
+    if (!isRecord(item) || typeof item.id !== 'string' || item.id.length === 0) continue;
+    ids.add(item.id);
+  }
+  return ids;
+}
+
+function acceptedTraceCreateEventIds(responseBody: string, requestBody: unknown): Set<string> {
+  const traceIds = traceCreateEventIds(requestBody);
+  if (traceIds.size === 0 || !responseBody) return traceIds;
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(responseBody);
+  } catch {
+    return traceIds;
+  }
+  if (!isRecord(parsed)) return traceIds;
+
+  const successes = eventResultIds(parsed.successes);
+  if (successes.size > 0) {
+    return new Set([...traceIds].filter((id) => successes.has(id)));
+  }
+
+  const errors = eventResultIds(parsed.errors);
+  if (errors.size > 0) {
+    return new Set([...traceIds].filter((id) => !errors.has(id)));
+  }
+
+  return traceIds;
+}
+
 function validateObjectBody(value: unknown): string | null {
   if (!isRecord(value)) return 'body must be a JSON object';
   if (typeof value.client_id !== 'string' || value.client_id.length === 0) {
@@ -470,16 +535,101 @@ function validateObjectBody(value: unknown): string | null {
   return null;
 }
 
+function manifestObjectsFromMetadata(metadata: unknown): ObjectUploadScopeObject[] {
+  if (!isRecord(metadata)) return [];
+  const out: ObjectUploadScopeObject[] = [];
+  for (const key of [
+    'attachment_manifest',
+    'artifact_manifest',
+    'input_text_snapshot_manifest',
+  ]) {
+    const entries = metadata[key];
+    if (!Array.isArray(entries)) continue;
+    for (const entry of entries) {
+      if (!isRecord(entry)) continue;
+      if (
+        typeof entry.storage_ref !== 'string' ||
+        !isAllowedObjectClass(entry.object_class) ||
+        typeof entry.size_bytes !== 'number' ||
+        !Number.isFinite(entry.size_bytes) ||
+        entry.size_bytes < 0 ||
+        typeof entry.sha256 !== 'string' ||
+        !/^sha256:[a-f0-9]{64}$/i.test(entry.sha256)
+      ) {
+        continue;
+      }
+      out.push({
+        storage_ref: entry.storage_ref,
+        object_class: entry.object_class,
+        size_bytes: Math.floor(entry.size_bytes),
+        sha256: entry.sha256.toLowerCase(),
+      });
+    }
+  }
+  return out;
+}
+
+async function registerObjectUploadScopes(
+  env: Env,
+  parsed: unknown,
+  acceptedTraceEventIds: Set<string>,
+): Promise<void> {
+  if (!env.TRACE_OBJECT_SCOPE_KV || !isRecord(parsed) || !Array.isArray(parsed.batch)) return;
+  for (const event of parsed.batch) {
+    if (!isRecord(event) || event.type !== 'trace-create' || !isRecord(event.body)) continue;
+    if (typeof event.id !== 'string' || !acceptedTraceEventIds.has(event.id)) continue;
+    const body = event.body;
+    const metadata = body.metadata;
+    if (!isRecord(metadata)) continue;
+    const clientId = typeof body.userId === 'string' && body.userId.length > 0
+      ? body.userId
+      : null;
+    const projectId = typeof metadata.projectId === 'string' && metadata.projectId.length > 0
+      ? metadata.projectId
+      : null;
+    const runId = typeof body.id === 'string' && body.id.length > 0
+      ? body.id
+      : null;
+    if (!clientId || !projectId || !runId) continue;
+    const objects = manifestObjectsFromMetadata(metadata);
+    if (objects.length === 0) continue;
+    await env.TRACE_OBJECT_SCOPE_KV.put(
+      objectAuthorityRegistryKey(clientId, projectId, runId),
+      JSON.stringify({ version: 1, client_id: clientId, project_id: projectId, run_id: runId, objects }),
+      { expirationTtl: OBJECT_SCOPE_TTL_SECONDS },
+    );
+  }
+}
+
+async function loadRegisteredObjectScopes(
+  env: Env,
+  clientId: string,
+  projectId: string,
+  runId: string,
+): Promise<ObjectUploadScopeObject[] | null> {
+  const raw = await env.TRACE_OBJECT_SCOPE_KV?.get(
+    objectAuthorityRegistryKey(clientId, projectId, runId),
+  );
+  if (!raw) return null;
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(raw);
+  } catch {
+    return null;
+  }
+  if (!isRecord(parsed) || !Array.isArray(parsed.objects)) return null;
+  const validationError = validateObjectScopeBody(parsed, OBJECT_AUTHORIZE_MAX_OBJECTS);
+  if (validationError != null) return null;
+  return (parsed as unknown as ObjectUploadTokenPayload).objects;
+}
+
 async function handleObjectAuthorizeRequest(request: Request, env: Env): Promise<Response> {
   if (request.headers.get(RELAY_MARKER_HEADER) !== OBJECT_RELAY_MARKER_VALUE) {
     return jsonResponse(403, { error: 'missing object client marker' });
   }
-  if (!hasObjectRelayConfig(env)) {
+  if (!hasObjectAuthorizeConfig(env)) {
     return jsonResponse(503, { error: 'object relay upload authority is not configured' });
   }
-  if (!allowTestObjectAuthorize(env)) {
-    return jsonResponse(403, { error: 'object upload authorization is disabled' });
-  }
   const contentType = request.headers.get('content-type') ?? '';
   if (!contentType.toLowerCase().includes('application/json')) {
     return jsonResponse(415, { error: 'content-type must be application/json' });
@@ -509,13 +659,38 @@ async function handleObjectAuthorizeRequest(request: Request, env: Env): Promise
     run_id: string;
     objects: ObjectUploadScopeObject[];
   };
+  const registeredObjects = await loadRegisteredObjectScopes(
+    env,
+    payload.client_id,
+    payload.project_id,
+    payload.run_id,
+  );
+  if (!registeredObjects) {
+    return jsonResponse(403, { error: 'object upload authority is not registered' });
+  }
+  const registeredByKey = new Map(
+    registeredObjects.map((object) => [objectScopeKey(object), object]),
+  );
+  for (const object of payload.objects) {
+    const registered = registeredByKey.get(objectScopeKey(object));
+    if (
+      !registered ||
+      registered.size_bytes !== object.size_bytes ||
+      registered.sha256.toLowerCase() !== object.sha256.toLowerCase()
+    ) {
+      return jsonResponse(403, { error: 'object upload authority scope mismatch' });
+    }
+  }
   const token = await signUploadToken(env, {
     version: 1,
     client_id: payload.client_id,
     project_id: payload.project_id,
     run_id: payload.run_id,
     exp: now + OBJECT_UPLOAD_TOKEN_TTL_SECONDS,
-    objects: payload.objects,
+    objects: payload.objects.map((object) => ({
+      ...object,
+      sha256: object.sha256.toLowerCase(),
+    })),
   });
   if (!token) return jsonResponse(503, { error: 'object relay upload authority is not configured' });
   return jsonResponse(200, {
@@ -659,7 +834,7 @@ async function handleRequest(request: Request, env: Env): Promise<Response> {
       ok: true,
       service: 'open-design-telemetry-relay',
       configured: hasLangfuseCredentials(env),
-      objectRelayConfigured: hasObjectRelayConfig(env),
+      objectRelayConfigured: hasObjectAuthorizeConfig(env),
       upstream: resolveLangfuseUrl(env),
     });
   }
@@ -718,6 +893,9 @@ async function handleRequest(request: Request, env: Env): Promise<Response> {
     body: rawBody,
   });
   const upstreamBody = await upstream.text();
+  if (upstream.ok) {
+    await registerObjectUploadScopes(env, parsed, acceptedTraceCreateEventIds(upstreamBody, parsed));
+  }
   return new Response(upstreamBody, {
     status: upstream.status,
     headers: {

apps/telemetry-worker/tests/index.test.ts

@@ -27,6 +27,16 @@ function makeRateLimiter(success: boolean) {
   };
 }
 
+function makeScopeKv(seed: Record<string, string> = {}) {
+  const values = new Map(Object.entries(seed));
+  return {
+    get: vi.fn(async (key: string) => values.get(key) ?? null),
+    put: vi.fn(async (key: string, value: string) => {
+      values.set(key, value);
+    }),
+  };
+}
+
 function objectRelayHeaders(): Record<string, string> {
   return {
     'Content-Type': 'application/json',
@@ -287,8 +297,9 @@ describe('telemetry worker', () => {
     fetchSpy.mockRestore();
   });
 
-  it('rejects public marker-only object authorization metadata', async () => {
+  it('rejects object authorization metadata without registered telemetry scope', async () => {
     const content = 'hello object';
+    const scopeKv = makeScopeKv();
     const response = await worker.fetch(
       new Request('https://telemetry.open-design.ai/api/objects/authorize', {
         method: 'POST',
@@ -315,17 +326,58 @@ describe('telemetry worker', () => {
         ...env,
         TRACE_OBJECT_BUCKET: { put: vi.fn() },
         TRACE_OBJECT_UPLOAD_SECRET: objectUploadSecret,
+        TRACE_OBJECT_SCOPE_KV: scopeKv,
       },
     );
 
     expect(response.status).toBe(403);
     expect(await response.json()).toEqual({
-      error: 'object upload authorization is disabled',
+      error: 'object upload authority is not registered',
     });
   });
 
-  it('issues test-only short-lived object upload tokens scoped to declared objects', async () => {
+  it('issues short-lived object upload tokens for scopes registered by telemetry traces', async () => {
     const content = 'hello object';
+    const scopeKv = makeScopeKv();
+    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(JSON.stringify({ successes: [], errors: [] }), { status: 207 }),
+    );
+    const traceResponse = await worker.fetch(
+      makeRequest({
+        batch: [
+          {
+            id: 'evt-1',
+            type: 'trace-create',
+            timestamp: '2026-06-08T00:00:00.000Z',
+            body: {
+              id: 'run-1',
+              name: 'open-design-turn',
+              userId: 'installation-1',
+              metadata: {
+                projectId: 'proj-1',
+                attachment_manifest: [
+                  {
+                    storage_ref: 'od://objects/workspaces/unknown/projects/proj-1/runs/run-1/attachment/att-1/brief.txt',
+                    object_class: 'attachment',
+                    size_bytes: content.length,
+                    sha256: `sha256:${requireSha256(content)}`,
+                  },
+                ],
+              },
+            },
+          },
+        ],
+      }),
+      {
+        ...env,
+        TRACE_OBJECT_BUCKET: { put: vi.fn() },
+        TRACE_OBJECT_UPLOAD_SECRET: objectUploadSecret,
+        TRACE_OBJECT_SCOPE_KV: scopeKv,
+      },
+    );
+    expect(traceResponse.status).toBe(207);
+    expect(scopeKv.put).toHaveBeenCalledTimes(1);
+
     const response = await worker.fetch(
       new Request('https://telemetry.open-design.ai/api/objects/authorize', {
         method: 'POST',
@@ -352,7 +404,7 @@ describe('telemetry worker', () => {
         ...env,
         TRACE_OBJECT_BUCKET: { put: vi.fn() },
         TRACE_OBJECT_UPLOAD_SECRET: objectUploadSecret,
-        TRACE_OBJECT_AUTHORIZE_TEST_ONLY: '1',
+        TRACE_OBJECT_SCOPE_KV: scopeKv,
       },
     );
 
@@ -360,6 +412,110 @@ describe('telemetry worker', () => {
     const body = await response.json() as { upload_token?: string; expires_at?: string };
     expect(body.upload_token).toEqual(expect.stringMatching(/^[A-Za-z0-9_-]+\.[a-f0-9]{64}$/));
     expect(body.expires_at).toEqual(expect.stringMatching(/^\d{4}-\d{2}-\d{2}T/));
+    fetchSpy.mockRestore();
+  });
+
+  it('registers object upload scopes when only a sibling observation is rejected', async () => {
+    const content = 'hello object';
+    const scopeKv = makeScopeKv();
+    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(
+        JSON.stringify({
+          successes: [{ id: 'trace-evt-1' }],
+          errors: [{ id: 'span-evt-1', status: 400 }],
+        }),
+        { status: 207 },
+      ),
+    );
+    const response = await worker.fetch(
+      makeRequest({
+        batch: [
+          {
+            id: 'trace-evt-1',
+            type: 'trace-create',
+            timestamp: '2026-06-08T00:00:00.000Z',
+            body: {
+              id: 'run-1',
+              name: 'open-design-turn',
+              userId: 'installation-1',
+              metadata: {
+                projectId: 'proj-1',
+                artifact_manifest: [
+                  {
+                    storage_ref: 'od://objects/workspaces/unknown/projects/proj-1/runs/run-1/artifact/art-1/index.html',
+                    object_class: 'artifact',
+                    size_bytes: content.length,
+                    sha256: `sha256:${requireSha256(content)}`,
+                  },
+                ],
+              },
+            },
+          },
+          {
+            id: 'span-evt-1',
+            type: 'span-create',
+            timestamp: '2026-06-08T00:00:00.000Z',
+            body: { id: 'span-1', traceId: 'run-1', name: 'agent-call' },
+          },
+        ],
+      }),
+      {
+        ...env,
+        TRACE_OBJECT_BUCKET: { put: vi.fn() },
+        TRACE_OBJECT_UPLOAD_SECRET: objectUploadSecret,
+        TRACE_OBJECT_SCOPE_KV: scopeKv,
+      },
+    );
+
+    expect(response.status).toBe(207);
+    expect(scopeKv.put).toHaveBeenCalledTimes(1);
+    fetchSpy.mockRestore();
+  });
+
+  it('does not register object upload scopes when Langfuse rejects the trace batch', async () => {
+    const scopeKv = makeScopeKv();
+    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(JSON.stringify({ successes: [], errors: [{ id: 'evt-1', status: 400 }] }), {
+        status: 207,
+      }),
+    );
+    const response = await worker.fetch(
+      makeRequest({
+        batch: [
+          {
+            id: 'evt-1',
+            type: 'trace-create',
+            timestamp: '2026-06-08T00:00:00.000Z',
+            body: {
+              id: 'run-1',
+              name: 'open-design-turn',
+              userId: 'installation-1',
+              metadata: {
+                projectId: 'proj-1',
+                artifact_manifest: [
+                  {
+                    storage_ref: 'od://objects/workspaces/unknown/projects/proj-1/runs/run-1/artifact/art-1/index.html',
+                    object_class: 'artifact',
+                    size_bytes: 12,
+                    sha256: `sha256:${requireSha256('hello object')}`,
+                  },
+                ],
+              },
+            },
+          },
+        ],
+      }),
+      {
+        ...env,
+        TRACE_OBJECT_BUCKET: { put: vi.fn() },
+        TRACE_OBJECT_UPLOAD_SECRET: objectUploadSecret,
+        TRACE_OBJECT_SCOPE_KV: scopeKv,
+      },
+    );
+
+    expect(response.status).toBe(207);
+    expect(scopeKv.put).not.toHaveBeenCalled();
+    fetchSpy.mockRestore();
   });
 
   it('rejects object batches without the object marker', async () => {