[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)

Author: Siri-Ray

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,6 +662,88 @@ describe('langfuse-bridge.reportRunCompletedFromDaemon', () => {
     );
   });
 
+  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 },
+    });
+    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.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',
@@ -695,6 +781,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 {
@@ -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"