582 lines
16 KiB
Go
582 lines
16 KiB
Go
package openai
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"iop/packages/go/streamgate"
|
|
)
|
|
|
|
// openAITunnelCodecState keeps caller-facing provider frames outside semantic
|
|
// evidence while the Core holds parsed endpoint events. It is reset before each
|
|
// recovery attempt, which is safe because path switches are allowed only before
|
|
// any response bytes are committed.
|
|
type openAITunnelCodecState struct {
|
|
mu sync.Mutex
|
|
endpoint string
|
|
releases [][]byte
|
|
terminal []byte
|
|
termSet bool
|
|
errorResponse *openAITunnelErrorResponse
|
|
}
|
|
|
|
type openAITunnelErrorResponse struct {
|
|
status int
|
|
headers map[string]string
|
|
body []byte
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) reset() {
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.releases = nil
|
|
s.terminal = nil
|
|
s.termSet = false
|
|
s.errorResponse = nil
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) bindEndpoint(endpoint string) bool {
|
|
if s == nil {
|
|
return false
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.endpoint == "" {
|
|
s.endpoint = endpoint
|
|
}
|
|
return s.endpoint == endpoint
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) endpointIsChat() bool {
|
|
if s == nil {
|
|
return false
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.endpoint == openAIRebuildEndpointChat
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) stageErrorResponseStart(status int, headers map[string]string) {
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.errorResponse = &openAITunnelErrorResponse{
|
|
status: status,
|
|
headers: cloneStringMap(headers),
|
|
}
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) appendErrorResponseWire(payload []byte) {
|
|
if s == nil || len(payload) == 0 {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
if s.errorResponse != nil {
|
|
s.errorResponse.body = append(s.errorResponse.body, payload...)
|
|
}
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) popErrorResponse() (openAITunnelErrorResponse, bool) {
|
|
if s == nil {
|
|
return openAITunnelErrorResponse{}, false
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.errorResponse == nil {
|
|
return openAITunnelErrorResponse{}, false
|
|
}
|
|
response := openAITunnelErrorResponse{
|
|
status: s.errorResponse.status,
|
|
headers: cloneStringMap(s.errorResponse.headers),
|
|
body: append([]byte(nil), s.errorResponse.body...),
|
|
}
|
|
s.errorResponse = nil
|
|
return response, true
|
|
}
|
|
|
|
func cloneStringMap(src map[string]string) map[string]string {
|
|
if len(src) == 0 {
|
|
return nil
|
|
}
|
|
dst := make(map[string]string, len(src))
|
|
for key, value := range src {
|
|
dst[key] = value
|
|
}
|
|
return dst
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) pushRelease(payload []byte) {
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.releases = append(s.releases, append([]byte(nil), payload...))
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) popRelease() ([]byte, bool) {
|
|
if s == nil {
|
|
return nil, false
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if len(s.releases) == 0 {
|
|
return nil, false
|
|
}
|
|
payload := s.releases[0]
|
|
s.releases = s.releases[1:]
|
|
return append([]byte(nil), payload...), true
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) setTerminal(payload []byte) {
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.terminal = append([]byte(nil), payload...)
|
|
s.termSet = true
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *openAITunnelCodecState) popTerminal() ([]byte, bool) {
|
|
if s == nil {
|
|
return nil, false
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if !s.termSet {
|
|
return nil, false
|
|
}
|
|
payload := append([]byte(nil), s.terminal...)
|
|
s.terminal = nil
|
|
s.termSet = false
|
|
return payload, true
|
|
}
|
|
|
|
// openAITunnelEndpointCodec parses Chat Completions or Responses SSE frames
|
|
// into semantic Core events while retaining each original frame for lossless
|
|
// downstream release.
|
|
type openAITunnelEndpointCodec struct {
|
|
endpoint string
|
|
state *openAITunnelCodecState
|
|
pending []byte
|
|
stagedWire []byte
|
|
chatTools map[int]openAITunnelToolIdentity
|
|
responseTools map[string]openAITunnelToolIdentity
|
|
terminal bool
|
|
}
|
|
|
|
type openAITunnelToolIdentity struct {
|
|
id string
|
|
name string
|
|
}
|
|
|
|
func newOpenAITunnelEndpointCodec(endpoint string, state *openAITunnelCodecState) *openAITunnelEndpointCodec {
|
|
if state == nil || (endpoint != openAIRebuildEndpointChat && endpoint != openAIRebuildEndpointResponses) {
|
|
return nil
|
|
}
|
|
if !state.bindEndpoint(endpoint) {
|
|
return nil
|
|
}
|
|
return &openAITunnelEndpointCodec{
|
|
endpoint: endpoint,
|
|
state: state,
|
|
chatTools: make(map[int]openAITunnelToolIdentity),
|
|
responseTools: make(map[string]openAITunnelToolIdentity),
|
|
}
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) decode(body []byte, flush bool) ([]streamgate.NormalizedEvent, error) {
|
|
if c == nil || c.terminal {
|
|
return nil, nil
|
|
}
|
|
c.pending = append(c.pending, body...)
|
|
var out []streamgate.NormalizedEvent
|
|
for {
|
|
frame, rest, ok := takeOpenAISSEFrame(c.pending)
|
|
if !ok {
|
|
break
|
|
}
|
|
c.pending = rest
|
|
events, err := c.decodeFrame(frame)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, events...)
|
|
if c.terminal {
|
|
c.pending = nil
|
|
return out, nil
|
|
}
|
|
}
|
|
if flush && len(c.pending) > 0 {
|
|
frame := append([]byte(nil), c.pending...)
|
|
c.pending = nil
|
|
events, err := c.decodeFrame(frame)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, events...)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func takeOpenAISSEFrame(buf []byte) (frame, rest []byte, ok bool) {
|
|
lf := bytes.Index(buf, []byte("\n\n"))
|
|
crlf := bytes.Index(buf, []byte("\r\n\r\n"))
|
|
end := -1
|
|
sepLen := 0
|
|
if lf >= 0 {
|
|
end, sepLen = lf, 2
|
|
}
|
|
if crlf >= 0 && (end < 0 || crlf < end) {
|
|
end, sepLen = crlf, 4
|
|
}
|
|
if end < 0 {
|
|
return nil, buf, false
|
|
}
|
|
frame = append([]byte(nil), buf[:end+sepLen]...)
|
|
rest = append([]byte(nil), buf[end+sepLen:]...)
|
|
return frame, rest, true
|
|
}
|
|
|
|
func openAISSEData(frame []byte) string {
|
|
normalized := strings.ReplaceAll(string(frame), "\r\n", "\n")
|
|
var lines []string
|
|
for _, line := range strings.Split(normalized, "\n") {
|
|
if !strings.HasPrefix(line, "data:") {
|
|
continue
|
|
}
|
|
lines = append(lines, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
|
|
}
|
|
return strings.Join(lines, "\n")
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) decodeFrame(frame []byte) ([]streamgate.NormalizedEvent, error) {
|
|
data := openAISSEData(frame)
|
|
if data == "" && json.Valid(bytes.TrimSpace(frame)) {
|
|
data = string(bytes.TrimSpace(frame))
|
|
}
|
|
if strings.TrimSpace(data) == "[DONE]" {
|
|
return c.finishTerminal(frame, false)
|
|
}
|
|
|
|
var events []streamgate.NormalizedEvent
|
|
var err error
|
|
if c.endpoint == openAIRebuildEndpointResponses {
|
|
events, err = c.decodeResponsesTunnelFrame(data)
|
|
} else {
|
|
events, err = c.decodeChatTunnelFrame(data)
|
|
}
|
|
if err == nil {
|
|
// Continue below.
|
|
} else {
|
|
return nil, err
|
|
}
|
|
|
|
releaseAttached := false
|
|
for _, event := range events {
|
|
switch event.Kind() {
|
|
case streamgate.EventKindTextDelta, streamgate.EventKindReasoningDelta, streamgate.EventKindToolCallFragment:
|
|
if releaseAttached == false {
|
|
payload := append(append([]byte(nil), c.stagedWire...), frame...)
|
|
c.stagedWire = nil
|
|
c.state.pushRelease(payload)
|
|
releaseAttached = true
|
|
} else {
|
|
c.state.pushRelease(nil)
|
|
}
|
|
case streamgate.EventKindProviderError:
|
|
if releaseAttached {
|
|
return c.finishTerminal(nil, true)
|
|
}
|
|
return c.finishTerminal(frame, true)
|
|
}
|
|
}
|
|
if releaseAttached == false {
|
|
// Provider opening/metadata and protocol-level finish frames are wire-only:
|
|
// retain them until the next semantic release or transport terminal rather
|
|
// than manufacturing text evidence from their JSON payload.
|
|
c.stagedWire = append(c.stagedWire, frame...)
|
|
}
|
|
return events, nil
|
|
}
|
|
|
|
// finishTransport turns the physical END boundary into the only terminal when
|
|
// no [DONE] marker already did so. A non-2xx response is a provider-error
|
|
// lifecycle event even if its body was opaque JSON and therefore wire-only.
|
|
func (c *openAITunnelEndpointCodec) finishTransport(body []byte, providerError bool) ([]streamgate.NormalizedEvent, error) {
|
|
if c == nil || c.terminal {
|
|
return nil, nil
|
|
}
|
|
events, err := c.decode(body, true)
|
|
if err == nil && c.terminal == false {
|
|
terminal, terminalErr := c.finishTerminal(nil, providerError)
|
|
if terminalErr != nil {
|
|
return nil, terminalErr
|
|
}
|
|
return append(events, terminal...), nil
|
|
}
|
|
return events, err
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) finishTerminal(frame []byte, providerError bool) ([]streamgate.NormalizedEvent, error) {
|
|
if c.terminal {
|
|
return nil, nil
|
|
}
|
|
payload := append(append([]byte(nil), c.stagedWire...), frame...)
|
|
c.stagedWire = nil
|
|
c.state.setTerminal(payload)
|
|
c.terminal = true
|
|
if providerError {
|
|
ev, err := newOpenAIProviderErrorEvent(streamGateErrorTunnelFailed)
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
}
|
|
ev, err := streamgate.NewTerminalEvent(streamGateChannelDefault, time.Now())
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) decodeChatTunnelFrame(data string) ([]streamgate.NormalizedEvent, error) {
|
|
if strings.TrimSpace(data) == "" {
|
|
return nil, nil
|
|
}
|
|
var payload struct {
|
|
Choices []struct {
|
|
Delta struct {
|
|
Content string `json:"content"`
|
|
Reasoning string `json:"reasoning"`
|
|
ReasoningContent string `json:"reasoning_content"`
|
|
ToolCalls []struct {
|
|
Index int `json:"index"`
|
|
ID string `json:"id"`
|
|
Function struct {
|
|
Name string `json:"name"`
|
|
Arguments string `json:"arguments"`
|
|
} `json:"function"`
|
|
} `json:"tool_calls"`
|
|
} `json:"delta"`
|
|
Message struct {
|
|
Content string `json:"content"`
|
|
ReasoningContent string `json:"reasoning_content"`
|
|
} `json:"message"`
|
|
FinishReason *string `json:"finish_reason"`
|
|
} `json:"choices"`
|
|
}
|
|
if err := json.Unmarshal([]byte(data), &payload); err != nil {
|
|
return nil, nil
|
|
}
|
|
var events []streamgate.NormalizedEvent
|
|
for _, choice := range payload.Choices {
|
|
content := choice.Delta.Content
|
|
if content == "" {
|
|
content = choice.Message.Content
|
|
}
|
|
if content != "" {
|
|
ev, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, content, time.Now())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
events = append(events, ev)
|
|
}
|
|
reasoning := choice.Delta.ReasoningContent
|
|
if reasoning == "" {
|
|
reasoning = choice.Delta.Reasoning
|
|
}
|
|
if reasoning == "" {
|
|
reasoning = choice.Message.ReasoningContent
|
|
}
|
|
if reasoning != "" {
|
|
ev, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, reasoning, time.Now())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
events = append(events, ev)
|
|
}
|
|
for _, tool := range choice.Delta.ToolCalls {
|
|
identity := c.chatTools[tool.Index]
|
|
if tool.ID != "" {
|
|
identity.id = tool.ID
|
|
}
|
|
if tool.Function.Name != "" {
|
|
identity.name = tool.Function.Name
|
|
}
|
|
c.chatTools[tool.Index] = identity
|
|
if tool.Function.Arguments == "" {
|
|
continue
|
|
}
|
|
id := identity.id
|
|
if id == "" {
|
|
id = fmt.Sprintf("tool-%d", tool.Index)
|
|
}
|
|
name := identity.name
|
|
if name == "" {
|
|
name = "function"
|
|
}
|
|
ev, err := streamgate.NewToolCallFragmentEvent(streamGateChannelDefault, id, name, tool.Function.Arguments, time.Now())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
events = append(events, ev)
|
|
}
|
|
// finish_reason is endpoint protocol state, not the transport terminal.
|
|
// Its frame remains in the release queue until [DONE] or END closes once.
|
|
}
|
|
return events, nil
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) rememberResponseTool(identity openAITunnelToolIdentity, keys ...string) {
|
|
if identity.id == "" && identity.name == "" {
|
|
return
|
|
}
|
|
for _, key := range keys {
|
|
if key == "" {
|
|
continue
|
|
}
|
|
current := c.responseTools[key]
|
|
if identity.id != "" {
|
|
current.id = identity.id
|
|
}
|
|
if identity.name != "" {
|
|
current.name = identity.name
|
|
}
|
|
c.responseTools[key] = current
|
|
}
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) responseTool(keys ...string) openAITunnelToolIdentity {
|
|
var identity openAITunnelToolIdentity
|
|
for _, key := range keys {
|
|
candidate := c.responseTools[key]
|
|
if identity.id == "" {
|
|
identity.id = candidate.id
|
|
}
|
|
if identity.name == "" {
|
|
identity.name = candidate.name
|
|
}
|
|
}
|
|
return identity
|
|
}
|
|
|
|
func (c *openAITunnelEndpointCodec) decodeResponsesTunnelFrame(data string) ([]streamgate.NormalizedEvent, error) {
|
|
if strings.TrimSpace(data) == "" {
|
|
return nil, nil
|
|
}
|
|
var payload struct {
|
|
Type string `json:"type"`
|
|
Delta string `json:"delta"`
|
|
ItemID string `json:"item_id"`
|
|
CallID string `json:"call_id"`
|
|
Name string `json:"name"`
|
|
OutputIdx int `json:"output_index"`
|
|
OutputText string `json:"output_text"`
|
|
Item struct {
|
|
ID string `json:"id"`
|
|
CallID string `json:"call_id"`
|
|
Name string `json:"name"`
|
|
} `json:"item"`
|
|
Output []struct {
|
|
Type string `json:"type"`
|
|
ID string `json:"id"`
|
|
CallID string `json:"call_id"`
|
|
Name string `json:"name"`
|
|
Arguments string `json:"arguments"`
|
|
Content []struct {
|
|
Type string `json:"type"`
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
} `json:"output"`
|
|
}
|
|
if err := json.Unmarshal([]byte(data), &payload); err != nil {
|
|
return nil, nil
|
|
}
|
|
outputKey := fmt.Sprintf("output-%d", payload.OutputIdx)
|
|
c.rememberResponseTool(openAITunnelToolIdentity{id: payload.CallID, name: payload.Name}, payload.CallID, payload.ItemID, outputKey)
|
|
c.rememberResponseTool(openAITunnelToolIdentity{id: payload.Item.CallID, name: payload.Item.Name}, payload.Item.CallID, payload.Item.ID, outputKey)
|
|
if payload.Type == "" && (payload.OutputText != "" || len(payload.Output) > 0) {
|
|
var events []streamgate.NormalizedEvent
|
|
text := payload.OutputText
|
|
if text == "" {
|
|
for _, item := range payload.Output {
|
|
for _, content := range item.Content {
|
|
if content.Type == "output_text" {
|
|
text += content.Text
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if text != "" {
|
|
event, eventErr := streamgate.NewTextDeltaEvent(streamGateChannelDefault, text, time.Now())
|
|
if eventErr != nil {
|
|
return nil, eventErr
|
|
}
|
|
events = append(events, event)
|
|
}
|
|
for i, item := range payload.Output {
|
|
key := fmt.Sprintf("output-%d", i)
|
|
c.rememberResponseTool(openAITunnelToolIdentity{id: item.CallID, name: item.Name}, item.CallID, item.ID, key)
|
|
if item.Type != "function_call" || item.Arguments == "" {
|
|
continue
|
|
}
|
|
identity := c.responseTool(item.CallID, item.ID, key)
|
|
id := identity.id
|
|
if id == "" {
|
|
id = fmt.Sprintf("call-%d", i)
|
|
}
|
|
name := identity.name
|
|
if name == "" {
|
|
name = "function"
|
|
}
|
|
event, eventErr := streamgate.NewToolCallFragmentEvent(streamGateChannelDefault, id, name, item.Arguments, time.Now())
|
|
if eventErr != nil {
|
|
return nil, eventErr
|
|
}
|
|
events = append(events, event)
|
|
}
|
|
return events, nil
|
|
}
|
|
switch payload.Type {
|
|
case "response.output_text.delta":
|
|
if payload.Delta == "" {
|
|
return nil, nil
|
|
}
|
|
ev, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, payload.Delta, time.Now())
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
case "response.reasoning_text.delta", "response.reasoning_summary_text.delta":
|
|
if payload.Delta == "" {
|
|
return nil, nil
|
|
}
|
|
ev, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, payload.Delta, time.Now())
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
case "response.function_call_arguments.delta":
|
|
if payload.Delta == "" {
|
|
return nil, nil
|
|
}
|
|
identity := c.responseTool(payload.CallID, payload.ItemID, outputKey)
|
|
id := identity.id
|
|
if id == "" {
|
|
id = fmt.Sprintf("call-%d", payload.OutputIdx)
|
|
}
|
|
name := identity.name
|
|
if name == "" {
|
|
name = "function"
|
|
}
|
|
ev, err := streamgate.NewToolCallFragmentEvent(streamGateChannelDefault, id, name, payload.Delta, time.Now())
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
case "response.completed", "response.incomplete":
|
|
return nil, nil
|
|
case "response.failed", "error":
|
|
ev, err := newOpenAIProviderErrorEvent(streamGateErrorTunnelFailed)
|
|
return []streamgate.NormalizedEvent{ev}, err
|
|
default:
|
|
return nil, nil
|
|
}
|
|
}
|