diff --git a/api/ApplyTrack.Api.Tests/OpenAiCompatibleLlmClientTests.cs b/api/ApplyTrack.Api.Tests/OpenAiCompatibleLlmClientTests.cs new file mode 100644 index 0000000..dfbeb37 --- /dev/null +++ b/api/ApplyTrack.Api.Tests/OpenAiCompatibleLlmClientTests.cs @@ -0,0 +1,128 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Aaron K. Clark + +using System.Net; +using System.Text; +using ApplyTrack.Api.Data; +using ApplyTrack.Api.Llm; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ApplyTrack.Api.Tests; + +public class OpenAiCompatibleLlmClientTests +{ + private static readonly EffectiveLlmConfig Cfg = + new("https://llm.example/v1", "test-model", "test-key", 30); + + [Fact] + public async Task Parses_a_normal_chat_completion() + { + var client = NewClient(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = Json("""{"choices":[{"message":{"content":" hello from model "}}]}"""), + }); + + var body = await client.CompleteAsync("system", "user", Cfg); + + Assert.Equal("hello from model", body); + } + + [Fact] + public async Task Rejects_success_response_when_declared_content_length_is_too_large() + { + var content = new StringContent("{}", Encoding.UTF8, "application/json"); + content.Headers.ContentLength = 1024 * 1024 + 1; + var client = NewClient(new HttpResponseMessage(HttpStatusCode.OK) { Content = content }); + + var ex = await Assert.ThrowsAsync(() => + client.CompleteAsync("system", "user", Cfg)); + + Assert.Contains("too large", ex.Message); + } + + [Fact] + public async Task Rejects_success_response_when_stream_exceeds_the_byte_cap() + { + var content = new StreamContent(new ChunkyStream(1024 * 1024 + 1)); + content.Headers.ContentType = new("application/json"); + var client = NewClient(new HttpResponseMessage(HttpStatusCode.OK) { Content = content }); + + var ex = await Assert.ThrowsAsync(() => + client.CompleteAsync("system", "user", Cfg)); + + Assert.Contains("too large", ex.Message); + } + + [Fact] + public async Task Oversized_error_body_is_not_read_into_memory() + { + var content = new StreamContent(new ChunkyStream(8 * 1024)); + content.Headers.ContentType = new("text/plain"); + var client = NewClient(new HttpResponseMessage(HttpStatusCode.BadGateway) { Content = content }); + + var ex = await Assert.ThrowsAsync(() => + client.CompleteAsync("system", "user", Cfg)); + + Assert.Equal("the LLM endpoint returned HTTP 502", ex.Message); + } + + private static OpenAiCompatibleLlmClient NewClient(HttpResponseMessage response) => + new(new SingleClientFactory(response), NullLogger.Instance); + + private static StringContent Json(string json) => + new(json, Encoding.UTF8, "application/json"); + + private sealed class SingleClientFactory(HttpResponseMessage response) : IHttpClientFactory + { + public HttpClient CreateClient(string name) => + new(new SingleResponseHandler(response)); + } + + private sealed class SingleResponseHandler(HttpResponseMessage response) : HttpMessageHandler + { + protected override Task SendAsync( + HttpRequestMessage request, CancellationToken cancellationToken) => + Task.FromResult(response); + } + + private sealed class ChunkyStream(long length) : Stream + { + private long _remaining = length; + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override int Read(byte[] buffer, int offset, int count) + { + if (_remaining <= 0) + return 0; + var read = (int)Math.Min(count, _remaining); + Array.Fill(buffer, (byte)'x', offset, read); + _remaining -= read; + return read; + } + + public override ValueTask ReadAsync( + Memory buffer, CancellationToken cancellationToken = default) + { + if (_remaining <= 0) + return ValueTask.FromResult(0); + var read = (int)Math.Min(buffer.Length, _remaining); + buffer[..read].Span.Fill((byte)'x'); + _remaining -= read; + return ValueTask.FromResult(read); + } + + public override void Flush() { } + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } +} diff --git a/api/ApplyTrack.Api/ApplyTrack.Api.csproj b/api/ApplyTrack.Api/ApplyTrack.Api.csproj index b86d987..65c099e 100644 --- a/api/ApplyTrack.Api/ApplyTrack.Api.csproj +++ b/api/ApplyTrack.Api/ApplyTrack.Api.csproj @@ -5,7 +5,7 @@ enable enable ApplyTrack.Api - 1.7.8 + 1.7.9 Aaron K. Clark Copyright 2026 Aaron K. Clark Apache-2.0 diff --git a/api/ApplyTrack.Api/Llm/OpenAiCompatibleLlmClient.cs b/api/ApplyTrack.Api/Llm/OpenAiCompatibleLlmClient.cs index 7c24e54..933f88c 100644 --- a/api/ApplyTrack.Api/Llm/OpenAiCompatibleLlmClient.cs +++ b/api/ApplyTrack.Api/Llm/OpenAiCompatibleLlmClient.cs @@ -5,6 +5,7 @@ using System.Net.Http.Headers; using System.Net.Http.Json; using System.Net.Sockets; +using System.Text; using System.Text.Json; using ApplyTrack.Api.Data; using ApplyTrack.Api.Scrape; @@ -20,6 +21,9 @@ namespace ApplyTrack.Api.Llm; /// public sealed class OpenAiCompatibleLlmClient : ILlmClient { + private const int MaxResponseBytes = 1024 * 1024; + private const int MaxErrorDetailBytes = 4 * 1024; + private readonly IHttpClientFactory _factory; private readonly ILogger _log; @@ -64,7 +68,7 @@ public async Task CompleteAsync( HttpResponseMessage res; try { - res = await http.SendAsync(req, ct); + res = await http.SendAsync(req, HttpCompletionOption.ResponseHeadersRead, ct); } catch (Exception ex) when (FindValidation(ex) is { } validation) { @@ -80,28 +84,31 @@ public async Task CompleteAsync( throw new LlmUnavailableException($"could not reach the LLM endpoint ({ex.Message})"); } - if (!res.IsSuccessStatusCode) + using (res) { - var detail = await SafeReadAsync(res, ct); - _log.LogWarning("LLM endpoint {Url} returned {Status}: {Detail}", url, (int)res.StatusCode, detail); - throw new LlmUnavailableException($"the LLM endpoint returned HTTP {(int)res.StatusCode}"); - } + if (!res.IsSuccessStatusCode) + { + var detail = await SafeReadAsync(res, ct); + _log.LogWarning("LLM endpoint {Url} returned {Status}: {Detail}", url, (int)res.StatusCode, detail); + throw new LlmUnavailableException($"the LLM endpoint returned HTTP {(int)res.StatusCode}"); + } - var json = await res.Content.ReadAsStringAsync(ct); - try - { - using var doc = JsonDocument.Parse(json); - var content = doc.RootElement - .GetProperty("choices")[0] - .GetProperty("message") - .GetProperty("content") - .GetString() ?? ""; - return content.Trim(); - } - catch (Exception ex) when (ex is JsonException or KeyNotFoundException or InvalidOperationException or IndexOutOfRangeException) - { - _log.LogWarning(ex, "Unexpected LLM response shape from {Url}", url); - throw new LlmUnavailableException("the LLM endpoint returned an unexpected response shape"); + var json = await ReadCappedStringAsync(res.Content, MaxResponseBytes, ct); + try + { + using var doc = JsonDocument.Parse(json); + var content = doc.RootElement + .GetProperty("choices")[0] + .GetProperty("message") + .GetProperty("content") + .GetString() ?? ""; + return content.Trim(); + } + catch (Exception ex) when (ex is JsonException or KeyNotFoundException or InvalidOperationException or IndexOutOfRangeException) + { + _log.LogWarning(ex, "Unexpected LLM response shape from {Url}", url); + throw new LlmUnavailableException("the LLM endpoint returned an unexpected response shape"); + } } } @@ -109,15 +116,43 @@ private static async Task SafeReadAsync(HttpResponseMessage res, Cancell { try { - var body = await res.Content.ReadAsStringAsync(ct); + var body = await ReadCappedStringAsync(res.Content, MaxErrorDetailBytes, ct); return body.Length > 500 ? body[..500] : body; } + catch (LlmUnavailableException) + { + return "(body too large)"; + } catch { return "(no body)"; } } + private static async Task ReadCappedStringAsync( + HttpContent content, int maxBytes, CancellationToken ct) + { + if (content.Headers.ContentLength is long declared && declared > maxBytes) + throw new LlmUnavailableException("the LLM endpoint returned a response that is too large"); + + await using var stream = await content.ReadAsStreamAsync(ct); + using var buffer = new MemoryStream(); + var chunk = new byte[16 * 1024]; + int read; + while ((read = await stream.ReadAsync(chunk, ct)) > 0) + { + if (buffer.Length + read > maxBytes) + throw new LlmUnavailableException("the LLM endpoint returned a response that is too large"); + buffer.Write(chunk, 0, read); + } + + var charset = content.Headers.ContentType?.CharSet?.Trim('"'); + Encoding encoding; + try { encoding = charset is null ? Encoding.UTF8 : Encoding.GetEncoding(charset); } + catch (ArgumentException) { encoding = Encoding.UTF8; } + return encoding.GetString(buffer.GetBuffer(), 0, (int)buffer.Length); + } + private static async ValueTask ConnectToPublicAddressOnlyAsync( SocketsHttpConnectionContext ctx, CancellationToken ct) { diff --git a/pyproject.toml b/pyproject.toml index 89ec55b..1759869 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "applytrack-poller" -version = "1.7.8" +version = "1.7.9" description = "Discovery poller for OSApplyTrack — fetches and scores remote job leads into shared Postgres." requires-python = ">=3.10" license = { text = "Apache-2.0" } diff --git a/uv.lock b/uv.lock index 89fdab3..28f48a0 100644 --- a/uv.lock +++ b/uv.lock @@ -23,7 +23,7 @@ wheels = [ [[package]] name = "applytrack-poller" -version = "1.7.8" +version = "1.7.9" source = { editable = "." } dependencies = [ { name = "defusedxml" },