protocol

package
v0.260806.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 6, 2026 License: MPL-2.0 Imports: 20 Imported by: 0

README

internal/protocol — request & response translation

This package owns the wire-level translation layer between the API styles tingly-box accepts from clients and the API styles it forwards to upstream providers. Every request that crosses the gateway flows through something in here.

Two ideas drive the design:

  • APIStyle — what a provider speaks. One of openai, anthropic, google. Discovered from provider config.
  • APIType — the canonical protocol shape we normalize a request to before forwarding. One of:
    • TypeOpenAIChat (/v1/chat/completions)
    • TypeOpenAIResponses (/v1/responses)
    • TypeAnthropicV1 (Anthropic Messages, the "stable" SDK type)
    • TypeAnthropicBeta (Anthropic Messages Beta, structural superset of v1)
    • TypeGoogle (Gemini generateContent)

A client speaks one APIType in, the dispatcher picks a target APIType based on the resolved provider's APIStyle, the transform chain converts the request, and a matching response-side converter turns the upstream reply back into the client's APIType.

Subpackage layout

Path Purpose
types.go, anthropic.go, openai.go, google.go shared types, helpers, and re-exports (the public canonical types live in ai/)
request/ request-side converters: <source>_to_<target>.go files that turn a parsed request of one APIType into another
transform/ the transform chain that wires request converters, consistency rules, and vendor quirks together; see below
stream/ streaming response converters — SSE in, SSE out, one file per (upstream type → client type) pair
nonstream/ non-streaming response converters — same pairing, JSON in, JSON out
assembler/ builders that turn a captured stream of upstream events into a single non-stream response object (used for recording, replay, and the "assemble then forward" path used by some Codex-style providers)
sse/ low-level SSE framing helpers shared by stream/
token/ token usage accounting helpers shared across converters
test/ shared test fixtures and golden data

internal/server/protocol_dispatch.go is the entrypoint that walks the (SourceAPI, TargetAPI) matrix and picks the right converter pair.

The transform chain

transform.NewTransformChain runs three transforms in order (see internal/server/chain_builder.go):

  1. BaseTransform (transform/base.go) — protocol conversion. Switches on targetType and calls the right request/ converter to reshape ctx.Request from the source APIType into the target APIType. This is the only step that changes the request's Go type.
  2. ConsistencyTransform (transform/consistency.go) — cross-provider normalization that applies to every provider of a given target type: tool schema cleanups (type: "object", properties shape), tool_call_id truncation, scenario flag application (disable-stream-usage, thinking mode), bounds checks.
  3. VendorTransform (transform/vendor.go) — provider-specific quirks keyed on the provider URL: DeepSeek's x_thinking → reasoning_content rewrite, Moonshot's reasoning shape, Codex's session-id handling, etc.

Optional pre/post recording transforms wrap the chain when scenario recording is enabled.

Supported (source → target) matrix

request/ and the response converters cover the following pairs. Anything not listed is intentionally unsupported and returns an unsupported request type error from BaseTransform.

source ↓ / target → openai_chat openai_responses anthropic_v1 anthropic_beta google
openai_chat ✓ (pass) ✓ — ✓ ✓
openai_responses ✓ ✓ (pass) — ✓ —
anthropic_v1 ✓ ✓ ✓ (pass) — ✓
anthropic_beta ✓ ✓ — ✓ (pass) ✓
google — — — — ✓ (pass)
  • ✓ (pass) means same-type passthrough — BaseTransform is a no-op and the request is forwarded as-is.
  • The — cells in the anthropic_v1 target column are the subject of the design concern below.
  • The harness validation matrix (internal/protocoltest.DefaultPairs) lists every supported pair explicitly. New rows here should be added there too so they get end-to-end coverage.

Design concern: Anthropic Beta is the single normalization target for non-Anthropic sources

Anthropic publishes two SDK shapes for the Messages API: stable v1 (anthropic.MessageNewParams) and beta (anthropic.BetaMessageNewParams). Beta is a structural superset of v1 — every v1 field exists on beta, and beta adds extras (extended thinking variants, more tool block types, MCP/server-tool blocks, etc.).

We used to normalize toward both targets, depending on the source:

  • OpenAIChat → Anthropic picked TypeAnthropicBeta.
  • OpenAIResponses → Anthropic picked TypeAnthropicV1.

That asymmetry forced the codebase to carry two parallel converter families for the same destination (a v1 family and a beta family of Convert*ToAnthropic*Request helpers, plus a Beta→V1 projection helper that did nothing but downshape Beta back to v1 so the asymmetric path could compile). The behavior was identical at the wire — Anthropic providers accept both — so the asymmetry was pure busywork.

The current rule is:

Non-Anthropic source → Anthropic provider always normalizes to TypeAnthropicBeta. TypeAnthropicV1 as a target exists only for Anthropic V1 → Anthropic V1 passthrough.

Concretely:

  • convertToAnthropicV1 in transform/base.go accepts only v1 input (passthrough) and v1-from-beta is rejected as incompatible. Anything else returns unsupported request type ... non-Anthropic sources must target Anthropic beta.
  • convertToAnthropicBeta is the single funnel for OpenAI Chat, OpenAI Responses, and (eventually) Google sources targeting an Anthropic provider.
  • The dispatch switch in internal/server/protocol_dispatch.go mirrors this: the TypeAnthropicV1 arm is a one-liner passthrough; the TypeAnthropicBeta arm fans out by SourceAPI to the right cross-format handler.
What this rules in and out
  • In: a single conversion path per (non-Anthropic source, Anthropic target) pair. Adding a new field handling means editing one converter, not two.
  • In: Beta-typed response shaping (buildResponsesPayloadFromAnthropicBeta, the Beta stream handler in stream/anthropic_beta_to_openai_responses.go) is the only Anthropic → non-Anthropic response converter we maintain.
  • Out: there is no path for "force a non-Anthropic request through v1 specifically". If a provider rejects a Beta-only field we'd need to either strip the field in ConsistencyTransform or add a Beta→Beta normalization step — not re-introduce a parallel v1 pipeline.

Adding a new conversion

  1. Pick a source and target APIType. Add a converter under request/ named <source>_to_<target>.go and wire it into the corresponding convertTo<Target> switch in transform/base.go.
  2. Add the response-side converter(s):
    • non-streaming under nonstream/,
    • streaming under stream/.
  3. If the dispatch matrix changes, update the inner switch in internal/server/protocol_dispatch.go for the new (SourceAPI, TargetAPI) case.
  4. Add a row to the matrix above and an entry to the e2e test in transform/e2e_test.go. Unsupported pairs should hit the default error branch in BaseTransform whose message matches the test's "unsupported request type" predicate, so the test will report them as NOT SUPPORTED automatically.
  5. Run go test ./internal/protocol/... and ./harness matrix (built from cli/harness) to confirm the conversion matrix is still green.

Documentation

Overview

Package protocol provides backward compatibility aliases to the public protocol package. All code should migrate to use "github.com/tingly-dev/tingly-box/protocol" directly.

Index

Constants

View Source 
const (
	APIStyleOpenAI    APIStyle = publicprotocol.APIStyleOpenAI
	APIStyleAnthropic APIStyle = publicprotocol.APIStyleAnthropic
	APIStyleGoogle    APIStyle = publicprotocol.APIStyleGoogle

	TypeOpenAIChat      APIType = publicprotocol.TypeOpenAIChat
	TypeOpenAIResponses APIType = publicprotocol.TypeOpenAIResponses
	TypeAnthropicV1     APIType = publicprotocol.TypeAnthropicV1
	TypeAnthropicBeta   APIType = publicprotocol.TypeAnthropicBeta
	TypeGoogle          APIType = publicprotocol.TypeGoogle

	CodexAPIBase = publicprotocol.CodexAPIBase
)

Re-export constants for backward compatibility

Variables

View Source 
var (
	NewTokenUsage          = publicprotocol.NewTokenUsage
	NewTokenUsageWithCache = publicprotocol.NewTokenUsageWithCache
	NewTokenUsageFull      = publicprotocol.NewTokenUsageFull
	ZeroTokenUsage         = publicprotocol.ZeroTokenUsage
)

Re-export functions for backward compatibility

Functions

func CommitFirstChunk added in v0.260611.1 

func CommitFirstChunk(c *gin.Context)

CommitFirstChunk signals a failover gate wrapping c.Writer (if any) that the first real stream chunk has been produced, so it flushes buffered output and switches to pass-through. No-op when no gate is installed. It does NOT record TTFT: the first byte is a structural event, not a content token. Each protocol marks the first content event itself.

func GetInputValue 

func GetInputValue(input responses.ResponseNewParamsInputUnion) any

GetInputValue extracts the raw input value from ResponseNewParamsInputUnion. Returns the underlying string, array, or nil.

func IsContextCanceled 

func IsContextCanceled(err error) bool

IsContextCanceled checks if the error is due to context cancellation.

func MarkFirstToken added in v0.260625.1 

func MarkFirstToken(c *gin.Context)

MarkFirstToken is the single source of truth for recording the first-token time used for TTFT: idempotent (earliest signal wins), safe to call repeatedly. Non-streaming handlers never call it, so their TTFT stays unset.

func PreprocessInputData added in v0.260409.1540 

func PreprocessInputData(data []byte) ([]byte, error)

PreprocessInputData preprocesses the JSON data before unmarshaling. It performs two preprocessing steps: 1. Adds "type": "message" to input items that don't have a type field 2. Flattens output_text content blocks into single strings

Items are inspected with gjson and only the ones that actually need a rewrite are re-serialized; untouched requests are returned as-is instead of being decoded and re-encoded item by item.

func RunLoop added in v0.260611.1 

func RunLoop(c *gin.Context, step func(w io.Writer) bool) bool

RunLoop drives a streaming response, handling client-disconnect detection, first-chunk commitment, and per-step flushing. It is the shared primitive used by both ProcessStream (typed-event hook pipeline) and raw-byte stream handlers.

step should write to w and return true to continue or false to stop. Returns true if the client disconnected mid-stream.

func SnapshotJSON added in v0.260806.1 

func SnapshotJSON(value any) ([]byte, error)

SnapshotJSON returns an owned JSON snapshot of one protocol value. SDK values may expose RawJSON to preserve fields unknown to their typed DTOs; typed nils are rejected before calling that method.

func UpstreamStatus added in v0.260611.1 

func UpstreamStatus(err error, fallback int) int

UpstreamStatus extracts the HTTP status code that an upstream provider returned, so the gateway can propagate it to the client instead of flattening every forwarding failure into a 500. It understands the error types returned by each vendor SDK (OpenAI / Anthropic share apierror.Error; google-genai uses genai.APIError). When the error does not carry a usable upstream status (e.g. a transport-level failure with no HTTP response), it returns fallback.

Types

type APIStyle 

type APIStyle = publicprotocol.APIStyle

Type aliases to public protocol types for backward compatibility

type APIType 

type APIType = publicprotocol.APIType

type AnthropicBetaMessagesRequest 

type AnthropicBetaMessagesRequest struct {
	Stream bool `json:"stream"`
	*anthropic.BetaMessageNewParams
}

Use official Anthropic SDK types directly

func (*AnthropicBetaMessagesRequest) MarshalJSON added in v0.260531.1 

func (r *AnthropicBetaMessagesRequest) MarshalJSON() ([]byte, error)

func (*AnthropicBetaMessagesRequest) UnmarshalJSON 

func (r *AnthropicBetaMessagesRequest) UnmarshalJSON(data []byte) error

type AnthropicMessagesRequest 

type AnthropicMessagesRequest struct {
	Stream bool `json:"stream"`
	*anthropic.MessageNewParams
}

Use official Anthropic SDK types directly

func (*AnthropicMessagesRequest) MarshalJSON added in v0.260531.1 

func (r *AnthropicMessagesRequest) MarshalJSON() ([]byte, error)

func (*AnthropicMessagesRequest) UnmarshalJSON 

func (r *AnthropicMessagesRequest) UnmarshalJSON(data []byte) error

type BetaRound 

type BetaRound struct {
	Messages       []anthropic.BetaMessageParam
	IsCurrentRound bool
	Stats          *RoundStats // Optional metadata about the round structure
}

BetaRound represents a conversation round for v1beta API.

type Client 

type Client = publicprotocol.Client

type ErrorDetail 

type ErrorDetail struct {
	Message string `json:"message"`
	Type    string `json:"type"`
	Code    string `json:"code,omitempty"`
}

ErrorDetail represents error details

type ErrorResponse 

type ErrorResponse struct {
	Error ErrorDetail `json:"error"`
}

ErrorResponse represents an error response

type GoogleRequest 

type GoogleRequest struct {
	Model    string
	Contents []*genai.Content
	Config   *genai.GenerateContentConfig
}

GoogleRequest wraps Google API request parameters Google's SDK uses separate parameters rather than a single request struct

type Grouper 

type Grouper struct{}

Grouper provides methods to group messages into conversation rounds.

func NewGrouper 

func NewGrouper() *Grouper

NewGrouper creates a new Grouper instance.

func (*Grouper) GroupBeta 

func (g *Grouper) GroupBeta(messages []anthropic.BetaMessageParam) []BetaRound

GroupBeta groups beta messages into conversation rounds.

func (*Grouper) GroupV1 

func (g *Grouper) GroupV1(messages []anthropic.MessageParam) []V1Round

GroupV1 groups v1 messages into conversation rounds. A round starts with a pure user message and includes all subsequent messages (assistant with tool use, tool results) until the next pure user message (exclusive).

func (*Grouper) IsPureBetaUserMessage 

func (g *Grouper) IsPureBetaUserMessage(msg anthropic.BetaMessageParam) bool

IsPureBetaUserMessage checks if a beta message is a pure user instruction.

func (*Grouper) IsPureUserMessage 

func (g *Grouper) IsPureUserMessage(msg anthropic.MessageParam) bool

IsPureUserMessage checks if a v1 message is a pure user instruction (not a tool result).

type GuardrailsBufferedEvent added in v0.260418.2200 

type GuardrailsBufferedEvent struct {
	EventType string
	Payload   map[string]interface{}
}

type GuardrailsStreamState added in v0.260418.2200 

type GuardrailsStreamState struct {
	// PendingBlockMessages stores early hook verdicts keyed by tool_use id.
	PendingBlockMessages map[string]string
	// PendingBlockedIndex tracks which content block index is currently blocked.
	PendingBlockedIndex map[int]string
	// RewroteBlockedToolUse is set once the current message's tool_use block has
	// been replaced by a synthetic guardrails text block. The subsequent
	// message_delta stop_reason must be rewritten away from tool_use.
	RewroteBlockedToolUse bool
	// AnthropicToolEvents buffers one tool_use block from start -> delta -> stop
	// so the rewrite layer can either flush the original events or replace them.
	AnthropicToolEvents map[int][]GuardrailsBufferedEvent
	// AnthropicToolIDs links the buffered block index back to the provider tool id.
	AnthropicToolIDs map[int]string
}

type HandleContext 

type HandleContext struct {
	// Gin context
	GinContext *gin.Context

	// Model info
	ResponseModel string

	// Guardrails runtime state shared across request/response/stream phases for
	// one proxied conversation.
	Guardrails *HandleGuardrails

	// Hooks for stream processing (chainable - multiple hooks can be added)
	OnStreamEventHooks    []func(event interface{}) error
	OnStreamCompleteHooks []func()
	OnStreamErrorHooks    []func(err error)

	// Stream configuration flags
	DisableStreamUsage bool // Don't include usage in streaming chunks

	// EstimatedInputTokens is a pre-computed input-token estimate used only as a
	// fallback when the upstream stream reports no usage. The caller computes it
	// and sets it here, so the stream handler depends on this scalar rather than
	// the request.
	EstimatedInputTokens int
}

HandleContext provides dependencies for handle functions. It uses the builder pattern for optional configuration and hooks.

func NewHandleContext 

func NewHandleContext(c *gin.Context, responseModel string) *HandleContext

NewHandleContext creates a new HandleContext with required dependencies.

func (*HandleContext) CallOnStreamComplete 

func (hc *HandleContext) CallOnStreamComplete()

CallOnStreamComplete calls all OnStreamComplete hooks. This is useful for non-streaming handlers that still need to invoke complete hooks.

func (*HandleContext) DispatchStreamError added in v0.260611.1 

func (hc *HandleContext) DispatchStreamError(err error)

func (*HandleContext) EnsureGuardrails added in v0.260418.2200 

func (hc *HandleContext) EnsureGuardrails() *HandleGuardrails

func (*HandleContext) EnsureGuardrailsStream added in v0.260418.2200 

func (hc *HandleContext) EnsureGuardrailsStream() *GuardrailsStreamState

func (*HandleContext) ProcessStream 

func (hc *HandleContext) ProcessStream(nextFunc func() (bool, error, interface{}), handleFunc func(interface{}) error) error

ProcessStream provides a generic framework for processing streaming responses. It handles context cancellation, error checking, and event processing. Internally delegates loop infrastructure to RunLoop.

nextFunc should return (true, nil, event) to continue, (false, nil, nil) to stop, or (false, err, nil) on error. handleFunc is called for each event after OnStreamEventHooks are invoked.

func (*HandleContext) ReleaseStreamState added in v0.260625.1 

func (hc *HandleContext) ReleaseStreamState()

ReleaseStreamState drops per-stream hooks and guardrail buffers after a stream has completed or errored. Hook closures often capture assemblers/recorders that aggregate response chunks; clearing them prevents protocol stream response state from staying reachable through a HandleContext after request completion.

func (*HandleContext) SendError 

func (hc *HandleContext) SendError(err error, errorType, code string)

SendError sends an error response to the client.

func (*HandleContext) SetupSSEHeaders 

func (hc *HandleContext) SetupSSEHeaders()

SetupSSEHeaders sets the standard SSE (Server-Sent Events) headers.

func (*HandleContext) WithOnStreamComplete 

func (hc *HandleContext) WithOnStreamComplete(hook func()) *HandleContext

WithOnStreamComplete adds a hook that is called when stream completes successfully. Multiple hooks can be added and will be called in order.

func (*HandleContext) WithOnStreamError 

func (hc *HandleContext) WithOnStreamError(hook func(error)) *HandleContext

WithOnStreamError adds a hook that is called when stream encounters an error. Multiple hooks can be added and will be called in order.

func (*HandleContext) WithOnStreamEvent 

func (hc *HandleContext) WithOnStreamEvent(hook func(interface{}) error) *HandleContext

WithOnStreamEvent adds a hook that is called for each stream event. Multiple hooks can be added and will be called in order.

type HandleGuardrails added in v0.260418.2200 

type HandleGuardrails struct {
	Enabled bool

	CredentialMask *guardrailscore.CredentialMaskState
	Stream         *GuardrailsStreamState
}

type OpenAIChatCompletionRequest 

type OpenAIChatCompletionRequest struct {
	*openai.ChatCompletionNewParams
	Stream bool `json:"stream"`
}

OpenAIChatCompletionRequest is a type alias for OpenAI chat completion request with extra fields.

func (*OpenAIChatCompletionRequest) MarshalJSON added in v0.260531.1 

func (r *OpenAIChatCompletionRequest) MarshalJSON() ([]byte, error)

func (*OpenAIChatCompletionRequest) UnmarshalJSON 

func (r *OpenAIChatCompletionRequest) UnmarshalJSON(data []byte) error

type OpenAIConfig 

type OpenAIConfig = publicprotocol.OpenAIConfig

type Response 

type Response = responses.Response

Response is an alias to the native OpenAI SDK type

type ResponseCreateRequest 

type ResponseCreateRequest struct {
	// Stream indicates whether to stream the response
	// This is not part of ResponseNewParams as streaming is controlled
	// by using NewStreaming() method on the SDK client
	Stream bool `json:"stream"`

	// Embed the native SDK type for all other fields
	*responses.ResponseNewParams
}

ResponseCreateRequest wraps the native ResponseNewParams with additional fields for proxy-specific handling like the `stream` parameter.

func (*ResponseCreateRequest) UnmarshalJSON 

func (r *ResponseCreateRequest) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for ResponseCreateRequest It handles both the custom Stream field and the embedded ResponseNewParams

type ResponseInputItemUnionParam 

type ResponseInputItemUnionParam = responses.ResponseInputItemUnionParam

ResponseInputItemUnionParam is an alias to the native OpenAI SDK type

type ResponseNewParams 

type ResponseNewParams = responses.ResponseNewParams

ResponseNewParams is an alias to the native OpenAI SDK type

type ResponseNewParamsInputUnion 

type ResponseNewParamsInputUnion = responses.ResponseNewParamsInputUnion

ResponseNewParamsInputUnion is an alias to the native OpenAI SDK type

type RoundStats 

type RoundStats struct {
	UserMessageCount int  // Number of pure user messages in this round (should be 1)
	AssistantCount   int  // Number of assistant messages
	ToolResultCount  int  // Number of tool result messages
	TotalMessages    int  // Total messages in the round
	HasThinking      bool // Whether any assistant message contains thinking blocks
}

RoundStats contains metadata about a round's message composition.

type TokenUsage 

type TokenUsage = publicprotocol.TokenUsage

type V1Round 

type V1Round struct {
	Messages       []anthropic.MessageParam
	IsCurrentRound bool
	Stats          *RoundStats // Optional metadata about the round structure
}

V1Round represents a conversation round for v1 API.

Source Files

View all Source files

Directories

Path Synopsis
streamemit
Package streamemit provides a decoupled emission layer on top of the Anthropic stream assemblers in internal/protocol/assembler.
 Package streamemit provides a decoupled emission layer on top of the Anthropic stream assemblers in internal/protocol/assembler.
Package stream — prime.go
 Package stream — prime.go
Package usage centralizes token extraction and normalization logic for all supported provider protocols.
 Package usage centralizes token extraction and normalization logic for all supported provider protocols.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.