lingmawire

README

lingma-protocol-go

Shared wire-protocol core for the Lingma / DeepSeek gateway family.

This module is the single implementation of the Lingma HTTP+SSE wire format and of the DeepSeek DSML tool markup that leaks through it. Three consumers share it, so a protocol fix lands once and reaches all of them:

lingma-protocol-go (this module, package lingmawire)
├── CLIProxyAPI  internal/translator/lingma          (native adapter)
├── CLIProxyAPI  provider-plugins/internal/lingma    (plugin adapter)
└── lingma-tap   internal/bridge                     (egress boundary)

What it owns

File	Responsibility
envelope.go	SSE data: prefix handling, double-JSON body unwrapping, [DONE]
error.go	Lingma error envelopes, details unwrapping, status mapping
messages.go	Agent ID inference, content: null normalization for tool calls
usage.go	Usage/token normalization across envelope revisions
events.go	Normalized stream event and tool-call types
dsml.go	DSML tag grammar, namespace tolerance, sentinel detection
argument.go	Parameter reassembly into tool arguments
sanitizer.go	Streaming DSML sanitizer: hold, recover, strip, diagnose

It deliberately does not own HTTP handlers, logging, retry policy, storage, auth, or downstream response shaping. Those stay in the adapters.

Why the sanitizer is streaming

The model sometimes writes its tool calls into the text channel instead of the structured tool_calls channel, using DeepSeek's private DSML format. A DSML tag is a tag name fenced by pipe runs:

< || DSML || parameter name="command" > ls -la < / || DSML || parameter >

Two details make this hard to catch with a regex:

• The pipes are fullwidth U+FF5C, not ASCII |. Spelled byte-for-byte, a closing tag is <, /, EF BD 9C, EF BD 9C, DSML, EF BD 9C, EF BD 9C, space, parameter, >. Any parser that searches for the ASCII spelling finds nothing and passes the markup through.
• The pipe count, the bar width, and the whitespace around the DSML keyword are transport noise rather than protocol semantics: hosts emit one bar per side or two, ASCII or fullwidth, and mix the two freely.

Because a stream can split at any byte, including inside the three-byte fullwidth bar and between an opening tag and its closing tag, sanitizing cannot be a single pass over complete text. A chunk that looks like ordinary text may only become recognisable as markup two chunks later, and by then the bytes have already been forwarded.

Sanitizer therefore holds back anything that could still turn out to be markup, and only releases text once the surrounding bytes prove it is text:

s := lingmawire.NewSanitizer(lingmawire.SanitizeOptions{
    DSMLMode:    lingmawire.DSMLRecover,
    Role:        lingmawire.RoleAssistant,
    ThoughtTags: true,
    KnownTools:  declaredTools, // optional, gates recovery
})
s.NoteNativeToolCalls(structuredCalls) // optional, dedupes
for _, delta := range deltas {
    events = append(events, s.Feed(delta)...)
}
events = append(events, s.Flush()...)
for _, d := range s.Diagnostics() { log.Printf("%s", d) }

For non-streaming content use SanitizeText. For a post-condition check use ResidualSanitizerCheck / ContainsDSMLSentinel.

Policy

• Complete and recoverable markup becomes a real tool call when DSMLMode is DSMLRecover, so the agent loop keeps running instead of stalling with an empty assistant turn.
• Fragments never become calls. A tool name with no arguments and no closing tag is withheld, because text that merely explains the protocol must not be able to execute a tool.
• Everything unrecognised is withheld and reported. Leaked markup must never reach assistant text: the next request replays it, the model imitates its own malformed output, and the degeneration compounds.
• Role matters. Tool markup belongs to assistant output. A tool result in user history is protocol, not pollution, and survives untouched.
• The tool channel is never rewritten. RoleTool content passes through byte for byte, so a tool that legitimately prints protocol text keeps doing so.

Testing

testdata/stream/ holds real fragments captured from a live gateway log, including the degraded shape that stalled sessions: no envelope, no call tag, the tool name riding in a parameter called name, and only the closing tag still carrying the DSML prefix.

TestLeakFixturesEverySplitPoint replays every fixture split at every byte offset and byte-by-byte, asserting that no markup reaches the text channel and that recovery is identical at every split.

Documentation

Index

• Constants
• func AgentID(modelName string) string
• func ContainsDSMLSentinel(s string) bool
• func HasToolCalls(value any) bool
• func IsAgentCommonModel(modelName string) bool
• func IsDone(raw []byte) bool
• func ModelConfigSource(modelName string) string
• func NormalizeAssistantToolCallContent(messages []map[string]any) []map[string]any
• func NormalizeUsage(usage gjson.Result) json.RawMessage
• func NormalizeUsageBytes(raw []byte) json.RawMessage
• func ResidualSanitizerCheck(sanitized string) bool
• func ResolveErrorDetails(outerMessage, outerType, outerCode string, details json.RawMessage) (string, string, string)
• func SanitizeText(content string, opts SanitizeOptions) (string, []ToolCallDelta, []Diagnostic)
• func StripSSEPrefix(raw []byte) []byte
• func UnwrapBody(raw []byte) []byte
• type DSMLMode
• type Diagnostic
  • func (d Diagnostic) String() string
• type DiagnosticKind
  • func (k DiagnosticKind) String() string
• type ErrorInfo
  • func ParseError(raw []byte) (*ErrorInfo, bool)
• type Role
• type SSEEvent
• type SanitizeOptions
• type Sanitizer
  • func NewSanitizer(opts SanitizeOptions) *Sanitizer
  • func (s *Sanitizer) Diagnostics() []Diagnostic
  • func (s *Sanitizer) Feed(chunk string) []SSEEvent
  • func (s *Sanitizer) Flush() []SSEEvent
  • func (s *Sanitizer) NoteNativeToolCalls(calls []ToolCallDelta)
• type TokenDetails
• type ToolCallDelta
• type Usage
  • func (u *Usage) Consolidate()
  • func (u *Usage) UnmarshalJSON(data []byte) error

Constants

const (
	AgentChat    = "agent_chat"
	AgentCommon  = "agent_common"
	SourceSystem = "system"
)

Functions

func AgentID

func AgentID(modelName string) string

AgentID returns the appropriate agent_id value for the given model name. Models that require agent_common get "agent_common"; all others get "agent_chat".

func ContainsDSMLSentinel

func ContainsDSMLSentinel(s string) bool

ContainsDSMLSentinel reports whether s carries any DSML pipe marker, whether or not it sits inside a well-formed tag.

Adapters use it as a cheap post-condition: sanitized content that still matches means a leak slipped through and should be counted and logged. The check never rewrites anything, so it is safe to run on output that has already been sanitized.

func HasToolCalls

func HasToolCalls(value any) bool

HasToolCalls reports whether value represents a non-empty list of tool calls.

func IsAgentCommonModel

func IsAgentCommonModel(modelName string) bool

IsAgentCommonModel reports whether the given model name requires the agent_common AgentId instead of agent_chat.

func IsDone

func IsDone(raw []byte) bool

IsDone checks if a Lingma SSE chunk indicates the end of the stream. It handles plain "[DONE]", "data: [DONE]", and the double-JSON envelope {"body":"[DONE]"}.

func ModelConfigSource

func ModelConfigSource(modelName string) string

ModelConfigSource returns the model_config.source value for the given model name. agent_chat models use "system" (enables reasoning); agent_common models use "".

func NormalizeAssistantToolCallContent

func NormalizeAssistantToolCallContent(messages []map[string]any) []map[string]any

NormalizeAssistantToolCallContent adapts OpenAI-compatible assistant tool-call history to the stricter provider behind Lingma. OpenAI permits content=null (or omitted content) when tool_calls is present, but that provider fails to recognize the tool_calls and then rejects the following tool result as orphaned. Setting content to "" preserves the message semantics while satisfying both schemas.

Only modifies messages satisfying all of: 1. role is "assistant" 2. tool_calls is non-empty 3. content is null or missing

Normal assistant, user, and tool messages with null content are preserved unchanged. The original slice and maps are not mutated in-place.

func NormalizeUsage

func NormalizeUsage(usage gjson.Result) json.RawMessage

NormalizeUsage converts Lingma usage payloads into canonical OpenAI Chat Completions usage format.

func NormalizeUsageBytes

func NormalizeUsageBytes(raw []byte) json.RawMessage

NormalizeUsageBytes parses raw JSON bytes and returns canonical OpenAI usage JSON.

func ResidualSanitizerCheck

func ResidualSanitizerCheck(sanitized string) bool

ResidualSanitizerCheck reports whether sanitized output still contains DSML markup. Adapters run it as a post-condition and count a hit as a defect: the sanitizer should have removed the markup, so a hit means a shape it does not yet know about reached the wire.

func ResolveErrorDetails

func ResolveErrorDetails(outerMessage, outerType, outerCode string, details json.RawMessage) (string, string, string)

ResolveErrorDetails inspects the details field of a Lingma provider error envelope. Details may be:

1. A JSON string encoding an inner error object: "{\"error\":{\"code\":...,\"message\":...,\"type\":...}}"
2. A raw JSON object: {"error":{...}} or {"code":...,"message":...}
3. A plain text string error message
4. Null or missing
5. Malformed JSON

It returns (resolvedMessage, resolvedType, resolvedCode). Inner error values take precedence over outer fields. If inner fields are missing, it falls back to the outer fields without losing errors.

func SanitizeText

func SanitizeText(content string, opts SanitizeOptions) (string, []ToolCallDelta, []Diagnostic)

SanitizeText is the byte-in/byte-out entry point for non-streaming content, such as rewriting a message body before it is stored or forwarded. Streaming callers should use a Sanitizer so a split tag is never emitted early.

It returns the cleaned text, every call recovered from markup, and the diagnostics describing what was found.

func StripSSEPrefix

func StripSSEPrefix(raw []byte) []byte

StripSSEPrefix strips the "data:" or "event:" prefix and surrounding whitespace from an SSE line. If the line is an "event:" line or empty, returns nil.

func UnwrapBody

func UnwrapBody(raw []byte) []byte

UnwrapBody unwraps a double-JSON envelope if present: e.g. {"body": "{\"choices\":...}"} or {"body": {...}} Returns the unwrapped inner payload bytes, or the trimmed bytes if not wrapped.

Types

type DSMLMode

type DSMLMode uint8

DSMLMode selects how leaked DeepSeek tool markup is handled.

The zero value is DSMLRecover on purpose: Go zero values must not disable the protection that exists to keep sessions from stalling, so a caller using SanitizeOptions{} gets the useful behaviour.

const (
	// DSMLRecover recovers a complete leaked call and strips everything else.
	DSMLRecover DSMLMode = iota
	// DSMLStrip removes leaked markup without synthesizing tool calls.
	DSMLStrip
	// DSMLDisabled passes content through untouched.
	DSMLDisabled
)

type Diagnostic

type Diagnostic struct {
	Kind   DiagnosticKind
	Origin tagOrigin
	Tool   string
	Text   string
}

Diagnostic is one observation about leaked or recovered markup.

func (Diagnostic) String

func (d Diagnostic) String() string

type DiagnosticKind

type DiagnosticKind uint8

DiagnosticKind classifies a sanitizer observation.

const (
	// DiagDroppedMarkup marks markup that was withheld without becoming a call.
	DiagDroppedMarkup DiagnosticKind = iota
	// DiagRecoveredCall marks a call reconstructed from leaked markup.
	DiagRecoveredCall
	// DiagUnknownTool marks a recoverable call whose tool was not declared.
	DiagUnknownTool
	// DiagDuplicateCall marks a recovered call the upstream already sent
	// natively, which is dropped to avoid executing it twice.
	DiagDuplicateCall
	// DiagResidualDSML marks DSML markup that survived sanitizing.
	DiagResidualDSML
)

func (DiagnosticKind) String

func (k DiagnosticKind) String() string

type ErrorInfo

type ErrorInfo struct {
	StatusCode int    `json:"status_code"`
	Message    string `json:"message"`
	Type       string `json:"type"`
	Code       string `json:"code,omitempty"`
}

ErrorInfo contains structured information about an error returned by Lingma upstream.

func ParseError

func ParseError(raw []byte) (*ErrorInfo, bool)

ParseError attempts to parse a Lingma SSE line or JSON response into an ErrorInfo. Returns (info, true) if an error was found, or (nil, false) if not an error.

type Role

type Role uint8

Role is the author of the content being sanitized. Tool markup only ever belongs to assistant output; a user or tool message that contains a tool result is protocol, not pollution, and must survive.

const (
	RoleAssistant Role = iota
	RoleUser
	RoleTool
)

type SSEEvent

type SSEEvent struct {
	// Type is "data", "finish", or "done".
	Type string
	// Content is the delta.content text (for text streaming).
	Content string
	// ReasoningContent is the delta.reasoning_content text (thinking/reasoning).
	ReasoningContent string
	// ToolCalls carries tool call deltas.
	ToolCalls []ToolCallDelta
	// FinishReason is set when the model finishes (e.g. "tool_calls", "stop").
	FinishReason string
	// Usage carries token usage (from finish/usage events).
	Usage *Usage
	// FirstTokenDuration is time-to-first-token in milliseconds from Lingma
	// finish metadata.
	FirstTokenDuration int
	// HasError indicates the event contains an error.
	HasError bool
	// ErrorMsg is the error message when HasError is true.
	ErrorMsg string
	// ErrorType is the error type when HasError is true.
	ErrorType string
	// ErrorCode is the upstream error code when present.
	ErrorCode string
	// Raw is the raw inner JSON payload.
	Raw []byte
}

SSEEvent is one normalized upstream stream event.

type SanitizeOptions

type SanitizeOptions struct {
	DSMLMode DSMLMode
	Role     Role
	// KnownTools, when non-nil, restricts recovery to tools the request
	// actually declared. A recovered name outside this set is stripped and
	// reported instead of being turned into a call the client cannot serve.
	KnownTools map[string]bool
	// ThoughtTags routes thought spans into ReasoningContent instead of
	// Content. Lingma wraps model reasoning this way.
	ThoughtTags bool
	// MaxHoldBytes bounds the buffer used to wait for a complete markup
	// region. A model that opens markup and never closes it should not make
	// the proxy hold an unbounded amount of text.
	MaxHoldBytes int
}

SanitizeOptions configures a Sanitizer. The zero value is the recommended configuration for assistant output.

type Sanitizer

type Sanitizer struct {
	// contains filtered or unexported fields
}

Sanitizer is a streaming filter for model tool markup. Feed it content deltas in order and call Flush when the stream ends. It is not safe for concurrent use.

Holding text back is the whole point: a chunk boundary can land anywhere, including inside the three-byte fullwidth bar and between an opening tag and its closing tag. Emitting early would push markup downstream that the rest of the stream later proves was protocol.

func NewSanitizer

func NewSanitizer(opts SanitizeOptions) *Sanitizer

NewSanitizer returns a Sanitizer for one message.

func (*Sanitizer) Diagnostics

func (s *Sanitizer) Diagnostics() []Diagnostic

Diagnostics returns everything recorded so far and clears the record.

func (*Sanitizer) Feed

func (s *Sanitizer) Feed(chunk string) []SSEEvent

Feed consumes one content delta and returns the events it completes.

func (*Sanitizer) Flush

func (s *Sanitizer) Flush() []SSEEvent

Flush finalizes the stream, recovering or discarding anything still held.

func (*Sanitizer) NoteNativeToolCalls

func (s *Sanitizer) NoteNativeToolCalls(calls []ToolCallDelta)

NoteNativeToolCalls registers the structured tool calls the upstream already sent for this message, so a recovered duplicate can be dropped instead of being executed twice.

type TokenDetails

type TokenDetails struct {
	CachedTokens    int `json:"cached_tokens,omitempty"`
	ReasoningTokens int `json:"reasoning_tokens,omitempty"`
}

TokenDetails holds nested token counters.

type ToolCallDelta

type ToolCallDelta struct {
	Index     int
	ID        string
	Name      string
	Arguments string
}

ToolCallDelta is a single streamed tool call.

type Usage

type Usage struct {
	PromptTokens     int `json:"prompt_tokens"`
	CompletionTokens int `json:"completion_tokens"`
	TotalTokens      int `json:"total_tokens"`
	// Aliyun/Lingma aliases.
	InputTokens  int `json:"input_tokens"`
	OutputTokens int `json:"output_tokens"`
	// Cached and reasoning tokens.
	CachedTokens    int `json:"cached_tokens,omitempty"`
	ReasoningTokens int `json:"reasoning_tokens,omitempty"`
	// Nested details extracted from the Lingma response.
	PromptTokensDetails     *TokenDetails `json:"prompt_tokens_details,omitempty"`
	CompletionTokensDetails *TokenDetails `json:"completion_tokens_details,omitempty"`
	InputTokensDetails      *TokenDetails `json:"input_tokens_details,omitempty"`
	OutputTokensDetails     *TokenDetails `json:"output_tokens_details,omitempty"`
}

Usage holds token accounting for one upstream response.

func (*Usage) Consolidate

func (u *Usage) Consolidate()

Consolidate fills the OpenAI aliases and hoists nested counters.

func (*Usage) UnmarshalJSON

func (u *Usage) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts the snake_case, camelCase, and provider-specific token counter names observed across Lingma envelope revisions.