Skip to content

revert(ai-gateway): remove temporary stream diagnostics #3976

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marius-kilocode wants to merge 1 commit into
base: main
Choose a base branch
from
+3 −770
Open

revert(ai-gateway): remove temporary stream diagnostics #3976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 3 additions & 45 deletions apps/web/src/app/api/openrouter/[...path]/route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,12 +97,6 @@ import {
hasMiddleOutTransform,
} from '@/lib/ai-gateway/providers/openrouter/request-helpers';
import { scheduleAutoRoutingMirror } from '@/lib/ai-gateway/auto-routing-mirror';
import {
createStreamLifecycleTracker,
observeEventStream,
shouldObserveEventStream,
STREAM_ATTEMPT_HEADER,
} from '@/lib/ai-gateway/o11y/stream-lifecycle.server';

export const maxDuration = 800;

Expand Down Expand Up @@ -720,8 +714,6 @@ export async function POST(request: NextRequest): Promise<NextResponseType<unkno
authContext: Promise.resolve({ organizationId }),
});

const observesProvider = effectiveProviderContext.provider.id === 'custom';
const attemptId = observesProvider ? crypto.randomUUID() : null;
const response = await upstreamRequest({
path,
search: url.search,
Expand Down Expand Up @@ -769,9 +761,7 @@ export async function POST(request: NextRequest): Promise<NextResponseType<unkno
});

// Return a service unavailable error instead of the 402
const errorResponse = temporarilyUnavailableResponse();
if (attemptId) errorResponse.headers.set(STREAM_ATTEMPT_HEADER, attemptId);
return errorResponse;
return temporarilyUnavailableResponse();
}

if (response.status >= 400) {
Expand All @@ -786,29 +776,7 @@ export async function POST(request: NextRequest): Promise<NextResponseType<unkno
});
}

let clonedReponse = response.clone(); // reading from body is side-effectful
const observesStream = shouldObserveEventStream({
provider_id: effectiveProviderContext.provider.id,
status: response.status,
has_body: response.body !== null,
content_type: response.headers.get('content-type'),
});
const streamTracker =
observesStream && attemptId
? createStreamLifecycleTracker({
attempt_id: attemptId,
provider_id: effectiveProviderContext.provider.id,
api_kind: requestBodyParsed.kind,
})
: null;
if (streamTracker && clonedReponse.body) {
const owner = clonedReponse;
const body = clonedReponse.body;
clonedReponse = new Response(
observeEventStream(body, outcome => streamTracker.observe('provider', outcome), owner),
owner
);
}
const clonedReponse = response.clone(); // reading from body is side-effectful

if (!shouldBlockOnClassify) {
classifyResult = await awaitClassifyAbuse(classifyPromise);
Expand Down Expand Up @@ -845,7 +813,6 @@ export async function POST(request: NextRequest): Promise<NextResponseType<unkno
isUserByok: !!effectiveProviderContext.userByok,
});
if (errorResponse) {
if (attemptId) errorResponse.headers.set(STREAM_ATTEMPT_HEADER, attemptId);
return errorResponse;
}
}
Expand All @@ -857,17 +824,8 @@ export async function POST(request: NextRequest): Promise<NextResponseType<unkno
requestBodyParsed.kind
);
if (rewrittenResponse) {
// Rewritten streams are excluded because their final bytes cannot be compared to the source.
if (attemptId) rewrittenResponse.headers.set(STREAM_ATTEMPT_HEADER, attemptId);
return rewrittenResponse;
}

const finalResponse = wrapInSafeNextResponse(response);
if (attemptId) finalResponse.headers.set(STREAM_ATTEMPT_HEADER, attemptId);
if (!streamTracker || !finalResponse.body) return finalResponse;

return new NextResponse(
observeEventStream(finalResponse.body, outcome => streamTracker.observe('final', outcome)),
finalResponse
);
return wrapInSafeNextResponse(response);
}
Loading