Fix streaming-only model requests

This commit is contained in:
Kaede Akino committed 2026-09-16 18:47:06 +08:00
1 parent de89c37aec
commit 6aa50bb896
4 files changed
+96 -3

No files matched your search

@@ -1,10 +1,96 @@
import AgentKit
import Foundation
import Synchronization
import TermindDomain
import Testing
@testable import TermindApplication
private nonisolated final class CommandGeneratorStreamingURLProtocol: URLProtocol, @unchecked Sendable {
struct State: Sendable {
var answer = ""
var requestCount = 0
var streamed = false
}
static let state = Mutex(State())
override class func canInit(with request: URLRequest) -> Bool {
request.url?.host == "command-generator.test"
}
override class func canonicalRequest(for request: URLRequest) -> URLRequest { request }
override func startLoading() {
var data = request.httpBody ?? Data()
if data.isEmpty, let stream = request.httpBodyStream {
stream.open()
defer { stream.close() }
var buffer = [UInt8](repeating: 0, count: 4096)
while stream.hasBytesAvailable {
let count = stream.read(&buffer, maxLength: buffer.count)
if count <= 0 { break }
data.append(contentsOf: buffer.prefix(count))
}
}
let body = (try? JSONSerialization.jsonObject(with: data)) as? [String: Any]
let streamed = body?["stream"] as? Bool == true
let answer = Self.state.withLock {
$0.requestCount += 1
$0.streamed = streamed
return $0.answer
}
let response = HTTPURLResponse(
url: request.url!, statusCode: streamed ? 200 : 400, httpVersion: "HTTP/1.1",
headerFields: ["Content-Type": "text/event-stream"]
)!
client?.urlProtocol(self, didReceive: response, cacheStoragePolicy: .notAllowed)
if streamed {
for fragment in [String(answer.prefix(answer.count / 2)), String(answer.dropFirst(answer.count / 2))] {
let chunk = try! JSONSerialization.data(withJSONObject: ["choices": [["delta": ["content": fragment]]]])
client?.urlProtocol(self, didLoad: Data("data: \(String(decoding: chunk, as: UTF8.self))\n\n".utf8))
}
client?.urlProtocol(
self, didLoad: Data("data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n".utf8)
)
} else {
client?.urlProtocol(self, didLoad: Data(#"{"detail":"Stream must be set to true"}"#.utf8))
}
client?.urlProtocolDidFinishLoading(self)
}
override func stopLoading() {}
}
@Suite(.serialized)
struct CommandGeneratorStreamingTests {
@Test(arguments: [#"{"command":"echo 你好"}"#, #"{"command":""}"#, "not JSON"])
func generatesAndValidatesACompleteCommandUsingStreamingTransport(_ answer: String) async throws {
CommandGeneratorStreamingURLProtocol.state.withLock { $0 = .init(answer: answer) }
let configuration = URLSessionConfiguration.ephemeral
configuration.protocolClasses = [CommandGeneratorStreamingURLProtocol.self]
let session = URLSession(configuration: configuration)
defer { session.invalidateAndCancel() }
let model = AgentProviderClient(
provider: ModelProvider(
name: "Streaming only", inferenceURL: "https://command-generator.test/v1/chat/completions"),
model: AIModel(id: "test"), secret: "test-secret", session: session
)
let context = CommandGeneratorContext(draft: "say hello", shell: "zsh", cwd: "/tmp")
if answer.contains("echo") {
let command = try await CommandGeneratorService().generate(context: context, model: model)
#expect(command == "echo 你好")
} else {
await #expect(throws: CommandGeneratorError.self) {
try await CommandGeneratorService().generate(context: context, model: model)
}
}
let state = CommandGeneratorStreamingURLProtocol.state.withLock { $0 }
#expect(state.streamed)
#expect(state.requestCount == 1)
}
}
private actor CommandGeneratorRequestRecorder {
private(set) var requests: [AgentModelRequest] = []
+7
View File
@@ -237,6 +237,13 @@ conversation pickers, and unknown purpose values safely default to chat.
Ordinary planning and acting runs have no turn or tool-call count ceiling; they end
when the model finishes, the user stops, a hook terminates, or an error occurs.
Guardian remains separately bounded so unavailable or inconclusive review fails closed.
Compact, conversation titles, and Command Generator use AgentKit's buffered-result
interface over streaming transport, not `stream: false` requests. They collect
validated results before applying them and never expose intermediate fragments.
Buffered SSE responses require a terminal marker and are capped at 16 MiB of raw
response bytes; cancellation, malformed data, and premature EOF discard partial
results. Ordinary agent streaming retains its existing execution policy, and
providers returning complete JSON remain supported.
Each ordinary model turn retries only transient provider transport/service failures
five times at 1/2/4/8/16 seconds. A retry reuses the logical message and request,
abandons its partial presentation, and cannot replay a tool because tool dispatch begins
+1 -1
View File
@@ -2595,7 +2595,7 @@
repositoryURL = "https://github.com/AkinoKaede/AgentKit.git";
requirement = {
kind = upToNextMajorVersion;
minimumVersion = 0.11.0;
minimumVersion = 0.11.1;
};
};
/* End XCRemoteSwiftPackageReference section */
@@ -6,8 +6,8 @@
"kind" : "remoteSourceControl",
"location" : "https://github.com/AkinoKaede/AgentKit.git",
"state" : {
"revision" : "8ac6e78bab3185a0a6fc9d1af1bfeb2f8db57ca9",
"version" : "0.11.0"
"revision" : "1721b56bdb66e5f18d358e4e93ff766552f44c8c",
"version" : "0.11.1"
}
},
{