toki
c90bb755a9
- Refactor plan, code-review, finalize-task-routing, refine-local-plans, router skills - Add agent-workflow-loop-orchestration skill and plan agent configs - Update roadmap: knowledge-tool-optimization milestones, stream-evidence-gate-core SDD - Add stream-evidence-gate-core task, archive, and Go streamgate package - Update dev-test inventory (edge/node smoke), agent-contract, edge-local-dev-guide - Deprecate USER_REVIEW for output-validation-filters SDD
1065 lines
36 KiB
Go
1065 lines
36 KiB
Go
package streamgate_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"iop/packages/go/streamgate"
|
|
)
|
|
|
|
// fakeOpenAIRawKind is an OpenAI-family-only raw event kind enum. It is a
|
|
// distinct Go type from streamgate.EventKind and from fakeAgentRawKind, so the
|
|
// OpenAI codec cannot accidentally consume a production kind or an agent-kind
|
|
// constant. Each constant is owned by the OpenAI converter alone.
|
|
type fakeOpenAIRawKind string
|
|
|
|
const (
|
|
rawOpenAIResponseStart fakeOpenAIRawKind = "openai_response_start"
|
|
rawOpenAITextDelta fakeOpenAIRawKind = "openai_text_delta"
|
|
rawOpenAIReasoningDelta fakeOpenAIRawKind = "openai_reasoning_delta"
|
|
rawOpenAIToolCallFragment fakeOpenAIRawKind = "openai_tool_call_fragment"
|
|
rawOpenAITerminal fakeOpenAIRawKind = "openai_terminal"
|
|
rawOpenAIProviderError fakeOpenAIRawKind = "openai_provider_error"
|
|
)
|
|
|
|
// fakeAgentRawKind is an agent-family-only raw event kind enum. It is a
|
|
// distinct Go type from streamgate.EventKind and from fakeOpenAIRawKind, so
|
|
// the agent codec cannot accidentally consume a production kind or an
|
|
// OpenAI-kind constant.
|
|
type fakeAgentRawKind string
|
|
|
|
const (
|
|
rawAgentResponseStart fakeAgentRawKind = "agent_response_start"
|
|
rawAgentTextDelta fakeAgentRawKind = "agent_text_delta"
|
|
rawAgentReasoningDelta fakeAgentRawKind = "agent_reasoning_delta"
|
|
rawAgentToolCallFragment fakeAgentRawKind = "agent_tool_call_fragment"
|
|
rawAgentTerminal fakeAgentRawKind = "agent_terminal"
|
|
rawAgentProviderError fakeAgentRawKind = "agent_provider_error"
|
|
)
|
|
|
|
// fakeOpenAIProviderFrame simulates an OpenAI provider tunnel codec by
|
|
// translating raw provider-frame inputs (status, text, reasoning, tool call,
|
|
// terminal, error) into normalized events and terminals. It is keyed by its
|
|
// own distinct fakeOpenAIRawKind enum, not by streamgate.EventKind. It does
|
|
// not import any apps/internal or proto types.
|
|
type fakeOpenAIProviderFrame struct {
|
|
kind fakeOpenAIRawKind
|
|
status int
|
|
headers map[string]string
|
|
text string
|
|
reasoning string
|
|
toolCallID string
|
|
toolCallName string
|
|
toolCallArgs string
|
|
errorType string
|
|
errorCode string
|
|
errorMessage string
|
|
errorParam string
|
|
}
|
|
|
|
// fakeAgentRuntimeEvent simulates an agent-family runtime event codec with
|
|
// its own distinct fakeAgentRawKind enum, but with runtime-event fields. It
|
|
// does not delegate to fakeOpenAIProviderFrame and does not import any
|
|
// apps/internal or proto types.
|
|
type fakeAgentRuntimeEvent struct {
|
|
kind fakeAgentRawKind
|
|
text string
|
|
reasoning string
|
|
toolCallID string
|
|
toolCallName string
|
|
toolCallArgs string
|
|
errorType string
|
|
errorCode string
|
|
errorMessage string
|
|
errorParam string
|
|
}
|
|
|
|
// normalizeOpenAIFrame converts a fakeOpenAIProviderFrame's raw inputs into a
|
|
// normalized event using production constructors. Each fakeOpenAIRawKind
|
|
// constant maps directly to its production constructor without calling any
|
|
// other converter.
|
|
func normalizeOpenAIFrame(f fakeOpenAIProviderFrame) (streamgate.NormalizedEvent, error) {
|
|
ts := time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)
|
|
switch f.kind {
|
|
case rawOpenAIResponseStart:
|
|
status := f.status
|
|
if status == 0 {
|
|
status = 200
|
|
}
|
|
headers := f.headers
|
|
if headers == nil {
|
|
headers = map[string]string{}
|
|
}
|
|
return streamgate.NewResponseStartEvent("ch-1", status, headers, ts)
|
|
case rawOpenAITextDelta:
|
|
return streamgate.NewTextDeltaEvent("ch-1", f.text, ts)
|
|
case rawOpenAIReasoningDelta:
|
|
return streamgate.NewReasoningDeltaEvent("ch-1", f.reasoning, ts)
|
|
case rawOpenAIToolCallFragment:
|
|
id := f.toolCallID
|
|
if id == "" {
|
|
id = "tc-1"
|
|
}
|
|
name := f.toolCallName
|
|
if name == "" {
|
|
name = "test"
|
|
}
|
|
args := f.toolCallArgs
|
|
if args == "" {
|
|
args = "{}"
|
|
}
|
|
return streamgate.NewToolCallFragmentEvent("ch-1", id, name, args, ts)
|
|
case rawOpenAITerminal:
|
|
return streamgate.NewTerminalEvent("ch-1", ts)
|
|
case rawOpenAIProviderError:
|
|
errorType := f.errorType
|
|
if errorType == "" {
|
|
errorType = "provider_error"
|
|
}
|
|
errorCode := f.errorCode
|
|
if errorCode == "" {
|
|
errorCode = "provider_error"
|
|
}
|
|
errorMessage := f.errorMessage
|
|
if errorMessage == "" {
|
|
errorMessage = "provider_error_message"
|
|
}
|
|
desc, err := streamgate.NewExternalDescriptor(errorType, errorCode, errorMessage, f.errorParam)
|
|
if err != nil {
|
|
return streamgate.NormalizedEvent{}, err
|
|
}
|
|
return streamgate.NewProviderErrorEvent("ch-1", desc, streamgate.FailureCauseChain{}, ts)
|
|
}
|
|
return streamgate.NormalizedEvent{}, errors.New("unknown openai raw kind")
|
|
}
|
|
|
|
// normalizeAgentEvent converts a fakeAgentRuntimeEvent's raw inputs into a
|
|
// normalized event using production constructors. Each fakeAgentRawKind
|
|
// constant maps directly to its production constructor without calling any
|
|
// other converter, ensuring the agent family codec is independently validated
|
|
// against the OpenAI provider frame.
|
|
func normalizeAgentEvent(f fakeAgentRuntimeEvent) (streamgate.NormalizedEvent, error) {
|
|
ts := time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)
|
|
switch f.kind {
|
|
case rawAgentResponseStart:
|
|
// Agent runtime events do not carry HTTP status/headers in their
|
|
// raw shape; this branch exists for matrix completeness.
|
|
return streamgate.NewResponseStartEvent("ch-1", 200, map[string]string{}, ts)
|
|
case rawAgentTextDelta:
|
|
return streamgate.NewTextDeltaEvent("ch-1", f.text, ts)
|
|
case rawAgentReasoningDelta:
|
|
return streamgate.NewReasoningDeltaEvent("ch-1", f.reasoning, ts)
|
|
case rawAgentToolCallFragment:
|
|
id := f.toolCallID
|
|
if id == "" {
|
|
id = "tc-1"
|
|
}
|
|
name := f.toolCallName
|
|
if name == "" {
|
|
name = "test"
|
|
}
|
|
args := f.toolCallArgs
|
|
if args == "" {
|
|
args = "{}"
|
|
}
|
|
return streamgate.NewToolCallFragmentEvent("ch-1", id, name, args, ts)
|
|
case rawAgentTerminal:
|
|
return streamgate.NewTerminalEvent("ch-1", ts)
|
|
case rawAgentProviderError:
|
|
errorType := f.errorType
|
|
if errorType == "" {
|
|
errorType = "agent_runtime_error"
|
|
}
|
|
errorCode := f.errorCode
|
|
if errorCode == "" {
|
|
errorCode = "agent_runtime_error"
|
|
}
|
|
errorMessage := f.errorMessage
|
|
if errorMessage == "" {
|
|
errorMessage = "agent_runtime_error_message"
|
|
}
|
|
desc, err := streamgate.NewExternalDescriptor(errorType, errorCode, errorMessage, f.errorParam)
|
|
if err != nil {
|
|
return streamgate.NormalizedEvent{}, err
|
|
}
|
|
return streamgate.NewProviderErrorEvent("ch-1", desc, streamgate.FailureCauseChain{}, ts)
|
|
}
|
|
return streamgate.NormalizedEvent{}, errors.New("unknown agent raw kind")
|
|
}
|
|
|
|
// fakeHostReleaseSink implements streamgate.ReleaseSink and captures exactly
|
|
// one terminal commit while rejecting subsequent attempts, enforcing single
|
|
// external terminal semantics in the test fixture.
|
|
type fakeHostReleaseSink struct {
|
|
committedTerminal streamgate.TerminalResult
|
|
committedCount int
|
|
outputBody string
|
|
state streamgate.CommitState
|
|
}
|
|
|
|
// Compile-time assertion that fakeHostReleaseSink implements ReleaseSink.
|
|
var _ streamgate.ReleaseSink = (*fakeHostReleaseSink)(nil)
|
|
|
|
func (s *fakeHostReleaseSink) CommitResponseStart(ctx context.Context, rs streamgate.ResponseStart) (streamgate.CommitState, error) {
|
|
s.state = streamgate.CommitStateStreamOpen
|
|
return s.state, nil
|
|
}
|
|
|
|
func (s *fakeHostReleaseSink) Release(ctx context.Context, ev streamgate.ReleaseEvent) (streamgate.CommitState, error) {
|
|
return s.state, nil
|
|
}
|
|
|
|
func (s *fakeHostReleaseSink) CommitTerminal(ctx context.Context, tr streamgate.TerminalResult) (streamgate.CommitState, error) {
|
|
if s.committedCount >= 1 {
|
|
return s.state, errors.New("streamgate test: second terminal rejected")
|
|
}
|
|
s.committedCount++
|
|
s.committedTerminal = tr
|
|
s.state = streamgate.CommitStateTerminalCommitted
|
|
// Record external JSON body via descriptor accessors only.
|
|
s.outputBody = formatErrorEnvelope(tr)
|
|
return s.state, nil
|
|
}
|
|
|
|
// formatErrorEnvelope produces a JSON error envelope from a TerminalResult
|
|
// using only the ExternalDescriptor accessor methods. It returns an empty
|
|
// string for success results.
|
|
func formatErrorEnvelope(tr streamgate.TerminalResult) string {
|
|
if !tr.Error() {
|
|
return ""
|
|
}
|
|
ed := tr.ExternalDesc()
|
|
if ed == nil {
|
|
return ""
|
|
}
|
|
env := struct {
|
|
Error struct {
|
|
Type string `json:"type"`
|
|
Message string `json:"message"`
|
|
} `json:"error"`
|
|
}{
|
|
Error: struct {
|
|
Type string `json:"type"`
|
|
Message string `json:"message"`
|
|
}{
|
|
Type: ed.Type(),
|
|
Message: ed.Message(),
|
|
},
|
|
}
|
|
b, _ := json.Marshal(env)
|
|
return string(b)
|
|
}
|
|
|
|
// encodeChatSSEError frames a TerminalResult as a Chat Completions SSE error
|
|
// response: one SSE data line with the JSON envelope followed by [DONE]. It
|
|
// uses only the ExternalDescriptor accessor methods for payload construction.
|
|
func encodeChatSSEError(tr streamgate.TerminalResult) (string, error) {
|
|
envelope := formatErrorEnvelope(tr)
|
|
if envelope == "" {
|
|
return "", nil
|
|
}
|
|
return "data: " + envelope + "\n\n" + "[DONE]\n\n", nil
|
|
}
|
|
|
|
// encodeResponsesJSONError frames a TerminalResult as a Responses API JSON
|
|
// error response: a single JSON envelope body. It uses only the
|
|
// ExternalDescriptor accessor methods for payload construction.
|
|
func encodeResponsesJSONError(tr streamgate.TerminalResult) (string, error) {
|
|
return formatErrorEnvelope(tr), nil
|
|
}
|
|
|
|
// expectedMatrix is a test-internal table that pairs each codec family's raw
|
|
// kind with the production EventKind, expected BaseEventDisposition, and
|
|
// expected typed accessor payload. The matrix is the single source of truth
|
|
// for the independent lifecycle assertion.
|
|
type expectedMatrix struct {
|
|
name string
|
|
openaiRaw fakeOpenAIRawKind
|
|
agentRaw fakeAgentRawKind
|
|
expectedKind streamgate.EventKind
|
|
expectedDisp streamgate.BaseEventDisposition
|
|
openaiStatus int
|
|
openaiHeaders map[string]string
|
|
openaiText string
|
|
openaiReason string
|
|
openaiTcID string
|
|
openaiTcName string
|
|
openaiTcArgs string
|
|
openaiErrType string
|
|
openaiErrCode string
|
|
openaiErrMsg string
|
|
agentText string
|
|
agentReason string
|
|
agentTcID string
|
|
agentTcName string
|
|
agentTcArgs string
|
|
agentErrType string
|
|
agentErrCode string
|
|
agentErrMsg string
|
|
}
|
|
|
|
// TestOpenAIAndAgentCodecContractMatrix verifies that the two codec families,
|
|
// each keyed by its own distinct raw kind enum and each with its own
|
|
// independent converter, produce NormalizedEvents whose actual production
|
|
// kind, Disposition(), Validate() result, and typed accessor payload all
|
|
// independently match the expectedMatrix. The OpenAI family uses
|
|
// fakeOpenAIRawKind; the agent family uses fakeAgentRawKind. Neither raw
|
|
// enum is streamgate.EventKind.
|
|
func TestOpenAIAndAgentCodecContractMatrix(t *testing.T) {
|
|
matrix := []expectedMatrix{
|
|
{
|
|
name: "response_start",
|
|
openaiRaw: rawOpenAIResponseStart,
|
|
agentRaw: rawAgentResponseStart,
|
|
expectedKind: streamgate.EventKindResponseStart,
|
|
expectedDisp: streamgate.BaseDispositionHold,
|
|
openaiStatus: 201,
|
|
openaiHeaders: map[string]string{"x-custom": "val"},
|
|
},
|
|
{
|
|
name: "text_delta",
|
|
openaiRaw: rawOpenAITextDelta,
|
|
agentRaw: rawAgentTextDelta,
|
|
expectedKind: streamgate.EventKindTextDelta,
|
|
expectedDisp: streamgate.BaseDispositionReleaseCandidate,
|
|
openaiText: "hello-from-openai",
|
|
agentText: "hello-from-agent",
|
|
},
|
|
{
|
|
name: "reasoning_delta",
|
|
openaiRaw: rawOpenAIReasoningDelta,
|
|
agentRaw: rawAgentReasoningDelta,
|
|
expectedKind: streamgate.EventKindReasoningDelta,
|
|
expectedDisp: streamgate.BaseDispositionReleaseCandidate,
|
|
openaiReason: "thinking-openai",
|
|
agentReason: "thinking-agent",
|
|
},
|
|
{
|
|
name: "tool_call",
|
|
openaiRaw: rawOpenAIToolCallFragment,
|
|
agentRaw: rawAgentToolCallFragment,
|
|
expectedKind: streamgate.EventKindToolCallFragment,
|
|
expectedDisp: streamgate.BaseDispositionReleaseCandidate,
|
|
openaiTcID: "tc-1",
|
|
openaiTcName: "get_weather",
|
|
openaiTcArgs: `{"q":"hi"}`,
|
|
agentTcID: "tc-2",
|
|
agentTcName: "search",
|
|
agentTcArgs: `{"q":"bye"}`,
|
|
},
|
|
{
|
|
name: "terminal",
|
|
openaiRaw: rawOpenAITerminal,
|
|
agentRaw: rawAgentTerminal,
|
|
expectedKind: streamgate.EventKindTerminal,
|
|
expectedDisp: streamgate.BaseDispositionTerminalSuccessCandidate,
|
|
},
|
|
{
|
|
name: "provider_error",
|
|
openaiRaw: rawOpenAIProviderError,
|
|
agentRaw: rawAgentProviderError,
|
|
expectedKind: streamgate.EventKindProviderError,
|
|
expectedDisp: streamgate.BaseDispositionTerminalErrorCandidate,
|
|
openaiErrType: "provider_timeout",
|
|
openaiErrCode: "provider_timeout",
|
|
openaiErrMsg: "provider_timed_out",
|
|
agentErrType: "agent_runtime_timeout",
|
|
agentErrCode: "agent_runtime_timeout",
|
|
agentErrMsg: "agent_timed_out",
|
|
},
|
|
}
|
|
|
|
for _, tc := range matrix {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
// --- OpenAI path ---
|
|
openaiFrame := fakeOpenAIProviderFrame{
|
|
kind: tc.openaiRaw,
|
|
text: tc.openaiText,
|
|
reasoning: tc.openaiReason,
|
|
toolCallID: tc.openaiTcID,
|
|
toolCallName: tc.openaiTcName,
|
|
toolCallArgs: tc.openaiTcArgs,
|
|
errorType: tc.openaiErrType,
|
|
errorCode: tc.openaiErrCode,
|
|
errorMessage: tc.openaiErrMsg,
|
|
}
|
|
if tc.name == "response_start" {
|
|
openaiFrame.status = tc.openaiStatus
|
|
openaiFrame.headers = tc.openaiHeaders
|
|
}
|
|
|
|
openaiEvent, err := normalizeOpenAIFrame(openaiFrame)
|
|
if err != nil {
|
|
t.Fatalf("openai normalize failed: %v", err)
|
|
}
|
|
|
|
// 1. Kind independently matches expected production kind.
|
|
if openaiEvent.Kind() != tc.expectedKind {
|
|
t.Errorf("openai Kind(): want %q, got %q", tc.expectedKind, openaiEvent.Kind())
|
|
}
|
|
|
|
// 2. Disposition() independently matches expected base disposition.
|
|
actualDisp, err := streamgate.BaseDispositionOf(openaiEvent.Kind())
|
|
if err != nil {
|
|
t.Fatalf("openai BaseDispositionOf(%q): %v", openaiEvent.Kind(), err)
|
|
}
|
|
if openaiEvent.Disposition() != tc.expectedDisp {
|
|
t.Errorf("openai Disposition(): want %q, got %q", tc.expectedDisp, openaiEvent.Disposition())
|
|
}
|
|
if actualDisp != tc.expectedDisp {
|
|
t.Errorf("openai BaseDispositionOf(kind) does not match matrix: want %q, got %q", tc.expectedDisp, actualDisp)
|
|
}
|
|
|
|
// 3. Validate() passes.
|
|
if err := openaiEvent.Validate(); err != nil {
|
|
t.Errorf("openai event Validate() = %v", err)
|
|
}
|
|
|
|
// 4. Kind-specific typed accessor payload independently verified.
|
|
verifyOpenAIPayload(t, openaiEvent, tc)
|
|
|
|
// --- Agent path ---
|
|
agentFrame := fakeAgentRuntimeEvent{
|
|
kind: tc.agentRaw,
|
|
text: tc.agentText,
|
|
reasoning: tc.agentReason,
|
|
toolCallID: tc.agentTcID,
|
|
toolCallName: tc.agentTcName,
|
|
toolCallArgs: tc.agentTcArgs,
|
|
errorType: tc.agentErrType,
|
|
errorCode: tc.agentErrCode,
|
|
errorMessage: tc.agentErrMsg,
|
|
}
|
|
|
|
agentEvent, err := normalizeAgentEvent(agentFrame)
|
|
if err != nil {
|
|
t.Fatalf("agent normalize failed: %v", err)
|
|
}
|
|
|
|
// 1. Kind independently matches expected production kind.
|
|
if agentEvent.Kind() != tc.expectedKind {
|
|
t.Errorf("agent Kind(): want %q, got %q", tc.expectedKind, agentEvent.Kind())
|
|
}
|
|
|
|
// 2. Disposition() independently matches expected base disposition.
|
|
agentActualDisp, err := streamgate.BaseDispositionOf(agentEvent.Kind())
|
|
if err != nil {
|
|
t.Fatalf("agent BaseDispositionOf(%q): %v", agentEvent.Kind(), err)
|
|
}
|
|
if agentEvent.Disposition() != tc.expectedDisp {
|
|
t.Errorf("agent Disposition(): want %q, got %q", tc.expectedDisp, agentEvent.Disposition())
|
|
}
|
|
if agentActualDisp != tc.expectedDisp {
|
|
t.Errorf("agent BaseDispositionOf(kind) does not match matrix: want %q, got %q", tc.expectedDisp, agentActualDisp)
|
|
}
|
|
|
|
// 3. Validate() passes.
|
|
if err := agentEvent.Validate(); err != nil {
|
|
t.Errorf("agent event Validate() = %v", err)
|
|
}
|
|
|
|
// 4. Kind-specific typed accessor payload independently verified.
|
|
verifyAgentPayload(t, agentEvent, tc)
|
|
})
|
|
}
|
|
|
|
// --- Cross-family independence: raw kinds are distinct types and cannot
|
|
// be swapped. Compile-time guarantee, not runtime. ---
|
|
var _ fakeOpenAIRawKind = rawOpenAIResponseStart
|
|
var _ fakeAgentRawKind = rawAgentResponseStart
|
|
// The following would not compile (and must not):
|
|
// var _ fakeOpenAIRawKind = rawAgentResponseStart
|
|
// var _ fakeAgentRawKind = rawOpenAIResponseStart
|
|
// var _ streamgate.EventKind = rawOpenAIResponseStart
|
|
// var _ streamgate.EventKind = rawAgentResponseStart
|
|
}
|
|
|
|
// verifyOpenAIPayload checks the OpenAI event's typed accessor payload against
|
|
// the expected matrix values, independently of the agent path.
|
|
func verifyOpenAIPayload(t *testing.T, ev streamgate.NormalizedEvent, tc expectedMatrix) {
|
|
t.Helper()
|
|
switch tc.expectedKind {
|
|
case streamgate.EventKindResponseStart:
|
|
rs, err := ev.AsResponseStart()
|
|
if err != nil {
|
|
t.Fatalf("openai AsResponseStart: %v", err)
|
|
}
|
|
if rs.Status() != tc.openaiStatus {
|
|
t.Errorf("openai ResponseStart.Status(): want %d, got %d", tc.openaiStatus, rs.Status())
|
|
}
|
|
for k, v := range tc.openaiHeaders {
|
|
if rs.Headers()[k] != v {
|
|
t.Errorf("openai ResponseStart.Headers()[%q] = %q, want %q", k, rs.Headers()[k], v)
|
|
}
|
|
}
|
|
case streamgate.EventKindTextDelta:
|
|
text, err := ev.AsTextDelta()
|
|
if err != nil {
|
|
t.Fatalf("openai AsTextDelta: %v", err)
|
|
}
|
|
if text != tc.openaiText {
|
|
t.Errorf("openai text delta: want %q, got %q", tc.openaiText, text)
|
|
}
|
|
case streamgate.EventKindReasoningDelta:
|
|
reason, err := ev.AsReasoningDelta()
|
|
if err != nil {
|
|
t.Fatalf("openai AsReasoningDelta: %v", err)
|
|
}
|
|
if reason != tc.openaiReason {
|
|
t.Errorf("openai reasoning delta: want %q, got %q", tc.openaiReason, reason)
|
|
}
|
|
case streamgate.EventKindToolCallFragment:
|
|
tcData, err := ev.AsToolCallFragment()
|
|
if err != nil {
|
|
t.Fatalf("openai AsToolCallFragment: %v", err)
|
|
}
|
|
if tcData.ID != tc.openaiTcID {
|
|
t.Errorf("openai tc ID: want %q, got %q", tc.openaiTcID, tcData.ID)
|
|
}
|
|
if tcData.Name != tc.openaiTcName {
|
|
t.Errorf("openai tc name: want %q, got %q", tc.openaiTcName, tcData.Name)
|
|
}
|
|
if tcData.Arguments != tc.openaiTcArgs {
|
|
t.Errorf("openai tc args: want %q, got %q", tc.openaiTcArgs, tcData.Arguments)
|
|
}
|
|
case streamgate.EventKindTerminal:
|
|
tr, err := ev.AsTerminal()
|
|
if err != nil {
|
|
t.Fatalf("openai AsTerminal: %v", err)
|
|
}
|
|
if !tr.Success() {
|
|
t.Error("openai terminal: expected success")
|
|
}
|
|
if tr.Error() {
|
|
t.Error("openai terminal: must not be error")
|
|
}
|
|
case streamgate.EventKindProviderError:
|
|
tr, err := ev.AsProviderError()
|
|
if err != nil {
|
|
t.Fatalf("openai AsProviderError: %v", err)
|
|
}
|
|
if !tr.Error() {
|
|
t.Error("openai provider error: expected error terminal")
|
|
}
|
|
if tr.Success() {
|
|
t.Error("openai provider error: must not be success")
|
|
}
|
|
ed := tr.ExternalDesc()
|
|
if ed == nil {
|
|
t.Fatal("openai provider error: descriptor is nil")
|
|
}
|
|
if ed.Type() != tc.openaiErrType {
|
|
t.Errorf("openai descriptor Type(): want %q, got %q", tc.openaiErrType, ed.Type())
|
|
}
|
|
if ed.Code() != tc.openaiErrCode {
|
|
t.Errorf("openai descriptor Code(): want %q, got %q", tc.openaiErrCode, ed.Code())
|
|
}
|
|
if ed.Message() != tc.openaiErrMsg {
|
|
t.Errorf("openai descriptor Message(): want %q, got %q", tc.openaiErrMsg, ed.Message())
|
|
}
|
|
}
|
|
}
|
|
|
|
// verifyAgentPayload checks the agent event's typed accessor payload against
|
|
// the expected matrix values, independently of the OpenAI path.
|
|
func verifyAgentPayload(t *testing.T, ev streamgate.NormalizedEvent, tc expectedMatrix) {
|
|
t.Helper()
|
|
switch tc.expectedKind {
|
|
case streamgate.EventKindResponseStart:
|
|
rs, err := ev.AsResponseStart()
|
|
if err != nil {
|
|
t.Fatalf("agent AsResponseStart: %v", err)
|
|
}
|
|
if rs.Status() != 200 {
|
|
t.Errorf("agent ResponseStart.Status(): want 200, got %d", rs.Status())
|
|
}
|
|
case streamgate.EventKindTextDelta:
|
|
text, err := ev.AsTextDelta()
|
|
if err != nil {
|
|
t.Fatalf("agent AsTextDelta: %v", err)
|
|
}
|
|
if text != tc.agentText {
|
|
t.Errorf("agent text delta: want %q, got %q", tc.agentText, text)
|
|
}
|
|
case streamgate.EventKindReasoningDelta:
|
|
reason, err := ev.AsReasoningDelta()
|
|
if err != nil {
|
|
t.Fatalf("agent AsReasoningDelta: %v", err)
|
|
}
|
|
if reason != tc.agentReason {
|
|
t.Errorf("agent reasoning delta: want %q, got %q", tc.agentReason, reason)
|
|
}
|
|
case streamgate.EventKindToolCallFragment:
|
|
tcData, err := ev.AsToolCallFragment()
|
|
if err != nil {
|
|
t.Fatalf("agent AsToolCallFragment: %v", err)
|
|
}
|
|
if tcData.ID != tc.agentTcID {
|
|
t.Errorf("agent tc ID: want %q, got %q", tc.agentTcID, tcData.ID)
|
|
}
|
|
if tcData.Name != tc.agentTcName {
|
|
t.Errorf("agent tc name: want %q, got %q", tc.agentTcName, tcData.Name)
|
|
}
|
|
if tcData.Arguments != tc.agentTcArgs {
|
|
t.Errorf("agent tc args: want %q, got %q", tc.agentTcArgs, tcData.Arguments)
|
|
}
|
|
case streamgate.EventKindTerminal:
|
|
tr, err := ev.AsTerminal()
|
|
if err != nil {
|
|
t.Fatalf("agent AsTerminal: %v", err)
|
|
}
|
|
if !tr.Success() {
|
|
t.Error("agent terminal: expected success")
|
|
}
|
|
if tr.Error() {
|
|
t.Error("agent terminal: must not be error")
|
|
}
|
|
case streamgate.EventKindProviderError:
|
|
tr, err := ev.AsProviderError()
|
|
if err != nil {
|
|
t.Fatalf("agent AsProviderError: %v", err)
|
|
}
|
|
if !tr.Error() {
|
|
t.Error("agent provider error: expected error terminal")
|
|
}
|
|
if tr.Success() {
|
|
t.Error("agent provider error: must not be success")
|
|
}
|
|
ed := tr.ExternalDesc()
|
|
if ed == nil {
|
|
t.Fatal("agent provider error: descriptor is nil")
|
|
}
|
|
if ed.Type() != tc.agentErrType {
|
|
t.Errorf("agent descriptor Type(): want %q, got %q", tc.agentErrType, ed.Type())
|
|
}
|
|
if ed.Code() != tc.agentErrCode {
|
|
t.Errorf("agent descriptor Code(): want %q, got %q", tc.agentErrCode, ed.Code())
|
|
}
|
|
if ed.Message() != tc.agentErrMsg {
|
|
t.Errorf("agent descriptor Message(): want %q, got %q", tc.agentErrMsg, ed.Message())
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestProviderErrorRetainsTerminalBaseDisposition verifies that provider error
|
|
// events retain terminal-error candidate disposition even when the codec
|
|
// families have distinct error fields. Both the OpenAI provider frame and
|
|
// agent runtime event produce a terminal-error candidate via their
|
|
// independent converters with their own distinct raw kind enums.
|
|
func TestProviderErrorRetainsTerminalBaseDisposition(t *testing.T) {
|
|
expectedDisp := streamgate.BaseDispositionTerminalErrorCandidate
|
|
|
|
openaiFrame := fakeOpenAIProviderFrame{
|
|
kind: rawOpenAIProviderError,
|
|
errorType: "provider_timeout",
|
|
errorCode: "provider_timeout",
|
|
errorMessage: "provider_timed_out",
|
|
}
|
|
agentEvent := fakeAgentRuntimeEvent{
|
|
kind: rawAgentProviderError,
|
|
errorType: "agent_runtime_timeout",
|
|
errorCode: "agent_runtime_timeout",
|
|
errorMessage: "agent_timed_out",
|
|
}
|
|
|
|
openaiEvent, err := normalizeOpenAIFrame(openaiFrame)
|
|
if err != nil {
|
|
t.Fatalf("openai normalize failed: %v", err)
|
|
}
|
|
|
|
agentEventResult, err := normalizeAgentEvent(agentEvent)
|
|
if err != nil {
|
|
t.Fatalf("agent normalize failed: %v", err)
|
|
}
|
|
|
|
// Kind check: each converter maps its own raw kind to the correct
|
|
// production kind.
|
|
kind := streamgate.EventKindProviderError
|
|
if openaiEvent.Kind() != kind {
|
|
t.Errorf("openai kind: want %v, got %v", kind, openaiEvent.Kind())
|
|
}
|
|
if agentEventResult.Kind() != kind {
|
|
t.Errorf("agent kind: want %v, got %v", kind, agentEventResult.Kind())
|
|
}
|
|
|
|
// Disposition check: each produced event's actual Disposition() matches
|
|
// the expected terminal-error candidate.
|
|
openaiDisp := openaiEvent.Disposition()
|
|
if openaiDisp != expectedDisp {
|
|
t.Errorf("openai Disposition(): want %v, got %v", expectedDisp, openaiDisp)
|
|
}
|
|
|
|
agentDisp := agentEventResult.Disposition()
|
|
if agentDisp != expectedDisp {
|
|
t.Errorf("agent Disposition(): want %v, got %v", expectedDisp, agentDisp)
|
|
}
|
|
|
|
// Validate() passes for each.
|
|
if err := openaiEvent.Validate(); err != nil {
|
|
t.Errorf("openai Validate(): %v", err)
|
|
}
|
|
if err := agentEventResult.Validate(); err != nil {
|
|
t.Errorf("agent Validate(): %v", err)
|
|
}
|
|
|
|
// AsProviderError yields an error TerminalResult for each.
|
|
openaiTr, err := openaiEvent.AsProviderError()
|
|
if err != nil {
|
|
t.Fatalf("openai AsProviderError: %v", err)
|
|
}
|
|
if !openaiTr.Error() {
|
|
t.Error("openai provider error terminal should be error type")
|
|
}
|
|
if openaiTr.Success() {
|
|
t.Error("openai provider error terminal must not be success")
|
|
}
|
|
|
|
agentTr, err := agentEventResult.AsProviderError()
|
|
if err != nil {
|
|
t.Fatalf("agent AsProviderError: %v", err)
|
|
}
|
|
if !agentTr.Error() {
|
|
t.Error("agent provider error terminal should be error type")
|
|
}
|
|
if agentTr.Success() {
|
|
t.Error("agent provider error terminal must not be success")
|
|
}
|
|
|
|
// Descriptor accessors return family-specific values independently.
|
|
openaiED := openaiTr.ExternalDesc()
|
|
agentED := agentTr.ExternalDesc()
|
|
if openaiED.Type() == agentED.Type() {
|
|
t.Errorf("openai and agent descriptor Type() should be distinct: both %q", openaiED.Type())
|
|
}
|
|
if openaiED.Message() == agentED.Message() {
|
|
t.Errorf("openai and agent descriptor Message() should be distinct: both %q", openaiED.Message())
|
|
}
|
|
}
|
|
|
|
// TestReleaseSinkCommitsSingleExternalTerminal verifies that the fake sink
|
|
// accepts exactly one terminal, rejects subsequent commits, and produces
|
|
// correct per-endpoint public framing through descriptor accessors only. The
|
|
// failure cause chain retains the latest four codes including both required
|
|
// codes ("repetition_loop_detected" and "recovery_notice_translation_failed"),
|
|
// proving the bound is functional, not suppressive.
|
|
func TestReleaseSinkCommitsSingleExternalTerminal(t *testing.T) {
|
|
ts := time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)
|
|
ctx := context.Background()
|
|
|
|
// --- Build a terminal with a bounded cause chain (5 inputs, latest 4) ---
|
|
desc, err := streamgate.NewExternalDescriptor("rate_limit_error", "rate_limit_exceeded", "rate_limit_exceeded", "")
|
|
if err != nil {
|
|
t.Fatalf("NewExternalDescriptor: %v", err)
|
|
}
|
|
|
|
chain := streamgate.FailureCauseChain{}
|
|
// Append 5 causes; only the latest 4 should be retained. Using 5 ensures
|
|
// "discarded_prelude" is dropped while both "repetition_loop_detected"
|
|
// and "recovery_notice_translation_failed" survive in the retained set,
|
|
// proving the bound is functional, not suppressive.
|
|
allCodes := []string{
|
|
"discarded_prelude",
|
|
"repetition_loop_detected",
|
|
"filter_blocked",
|
|
"validation_failed",
|
|
"recovery_notice_translation_failed",
|
|
}
|
|
for _, code := range allCodes {
|
|
cause, err := streamgate.NewFailureCause("stage", code, "consumer-1", "filter-1", "")
|
|
if err != nil {
|
|
t.Fatalf("NewFailureCause(%s): %v", code, err)
|
|
}
|
|
chain, err = chain.Append(cause)
|
|
if err != nil {
|
|
t.Fatalf("chain.Append(%s): %v", code, err)
|
|
}
|
|
}
|
|
|
|
// Chain must retain exactly 4 causes (latest 4).
|
|
if chain.Len() != 4 {
|
|
t.Fatalf("chain Len() = %d, want 4", chain.Len())
|
|
}
|
|
|
|
// Both required codes must be present in the retained chain, in exact
|
|
// retained order, proving the bound is functional (latest 4 retained),
|
|
// not suppressive. The dropped prelude must not be present.
|
|
expectedCodes := []string{
|
|
"repetition_loop_detected",
|
|
"filter_blocked",
|
|
"validation_failed",
|
|
"recovery_notice_translation_failed",
|
|
}
|
|
retained := chain.All()
|
|
if len(retained) != len(expectedCodes) {
|
|
t.Fatalf("retained chain length = %d, want %d", len(retained), len(expectedCodes))
|
|
}
|
|
for i, want := range expectedCodes {
|
|
if got := retained[i].Code(); got != want {
|
|
t.Errorf("retained[%d].Code() = %q, want %q", i, got, want)
|
|
}
|
|
}
|
|
// The dropped code must not be present in the retained chain.
|
|
hasPrelude := false
|
|
for _, c := range retained {
|
|
if c.Code() == "discarded_prelude" {
|
|
hasPrelude = true
|
|
break
|
|
}
|
|
}
|
|
if hasPrelude {
|
|
t.Error("retained chain must not include 'discarded_prelude' (should be dropped)")
|
|
}
|
|
|
|
tr, err := streamgate.NewErrorTerminalResult("ch-1", desc, chain, ts)
|
|
if err != nil {
|
|
t.Fatalf("NewErrorTerminalResult: %v", err)
|
|
}
|
|
|
|
// --- Chat SSE endpoint subtest ---
|
|
t.Run("chat_sse_endpoint", func(t *testing.T) {
|
|
sink := &fakeHostReleaseSink{}
|
|
|
|
rs, err := streamgate.NewResponseStart("ch-1", 200, map[string]string{}, ts)
|
|
if err != nil {
|
|
t.Fatalf("NewResponseStart: %v", err)
|
|
}
|
|
_, err = sink.CommitResponseStart(ctx, rs)
|
|
if err != nil {
|
|
t.Fatalf("CommitResponseStart: %v", err)
|
|
}
|
|
|
|
// First commit should succeed.
|
|
state, err := sink.CommitTerminal(ctx, tr)
|
|
if err != nil {
|
|
t.Fatalf("first CommitTerminal failed: %v", err)
|
|
}
|
|
if state != streamgate.CommitStateTerminalCommitted {
|
|
t.Errorf("first commit state: want terminal_committed, got %v", state)
|
|
}
|
|
if sink.committedCount != 1 {
|
|
t.Errorf("committedCount = %d, want 1", sink.committedCount)
|
|
}
|
|
|
|
// Verify Chat SSE framing: data: {envelope}\n\n[DONE]\n\n
|
|
sseOutput, err := encodeChatSSEError(sink.committedTerminal)
|
|
if err != nil {
|
|
t.Fatalf("encodeChatSSEError: %v", err)
|
|
}
|
|
|
|
expectedEnvelope := formatErrorEnvelope(sink.committedTerminal)
|
|
expectedSSE := "data: " + expectedEnvelope + "\n\n" + "[DONE]\n\n"
|
|
if sseOutput != expectedSSE {
|
|
t.Errorf("Chat SSE output:\n got: %q\nwant: %q", sseOutput, expectedSSE)
|
|
}
|
|
|
|
// SSE output must start with "data: ".
|
|
if !strings.HasPrefix(sseOutput, "data: ") {
|
|
t.Errorf("Chat SSE must start with 'data: ', got: %q", sseOutput)
|
|
}
|
|
|
|
// SSE output must contain "[DONE]" terminator.
|
|
if !strings.Contains(sseOutput, "[DONE]") {
|
|
t.Errorf("Chat SSE must contain '[DONE]', got: %q", sseOutput)
|
|
}
|
|
|
|
// Decode the envelope and verify exact key set.
|
|
var decoded map[string]interface{}
|
|
if err := json.Unmarshal([]byte(expectedEnvelope), &decoded); err != nil {
|
|
t.Fatalf("envelope JSON decode failed: %v; body=%q", err, expectedEnvelope)
|
|
}
|
|
|
|
if _, ok := decoded["error"]; !ok {
|
|
t.Errorf("decoded top-level keys: missing 'error'; got %+v", decoded)
|
|
}
|
|
topKeys := make([]string, 0, len(decoded))
|
|
for k := range decoded {
|
|
topKeys = append(topKeys, k)
|
|
}
|
|
if len(topKeys) != 1 || topKeys[0] != "error" {
|
|
t.Errorf("decoded top-level keys: want ['error'], got %v", topKeys)
|
|
}
|
|
|
|
errObj, ok := decoded["error"].(map[string]interface{})
|
|
if !ok {
|
|
t.Fatalf("decoded['error'] is not object; got %T", decoded["error"])
|
|
}
|
|
errKeys := make([]string, 0, len(errObj))
|
|
for k := range errObj {
|
|
errKeys = append(errKeys, k)
|
|
}
|
|
if len(errKeys) != 2 {
|
|
t.Errorf("decoded['error'] keys: want 2, got %d (%v)", len(errKeys), errKeys)
|
|
}
|
|
if _, ok := errObj["type"]; !ok {
|
|
t.Error("decoded['error'] missing 'type'")
|
|
}
|
|
if _, ok := errObj["message"]; !ok {
|
|
t.Error("decoded['error'] missing 'message'")
|
|
}
|
|
if errObj["type"] != "rate_limit_error" {
|
|
t.Errorf("decoded['error'].type = %v, want rate_limit_error", errObj["type"])
|
|
}
|
|
if errObj["message"] != "rate_limit_exceeded" {
|
|
t.Errorf("decoded['error'].message = %v, want 'rate_limit_exceeded'", errObj["message"])
|
|
}
|
|
|
|
// Forbidden internal fields must not appear anywhere in the SSE output.
|
|
forbiddenFields := []string{"causes", "stack", "trace", "provider", "internal", "raw", "repetition_loop", "filter_blocked", "validation_failed", "cache_miss", "stage", "consumer"}
|
|
for _, f := range forbiddenFields {
|
|
if strings.Contains(sseOutput, f) {
|
|
t.Errorf("forbidden field '%s' found in Chat SSE output: %s", f, sseOutput)
|
|
}
|
|
}
|
|
|
|
// Verify the committed terminal reflects the bounded cause chain.
|
|
committed := sink.committedTerminal
|
|
if !committed.Error() {
|
|
t.Error("committed terminal should be error type")
|
|
}
|
|
committedCauses := committed.FailureCauses()
|
|
if committedCauses.Len() != 4 {
|
|
t.Errorf("committed causes Len() = %d, want 4", committedCauses.Len())
|
|
}
|
|
})
|
|
|
|
// --- Responses JSON endpoint subtest ---
|
|
t.Run("responses_json_endpoint", func(t *testing.T) {
|
|
sink := &fakeHostReleaseSink{}
|
|
|
|
rs, err := streamgate.NewResponseStart("ch-1", 200, map[string]string{}, ts)
|
|
if err != nil {
|
|
t.Fatalf("NewResponseStart: %v", err)
|
|
}
|
|
_, err = sink.CommitResponseStart(ctx, rs)
|
|
if err != nil {
|
|
t.Fatalf("CommitResponseStart: %v", err)
|
|
}
|
|
|
|
// First commit should succeed.
|
|
state, err := sink.CommitTerminal(ctx, tr)
|
|
if err != nil {
|
|
t.Fatalf("first CommitTerminal failed: %v", err)
|
|
}
|
|
if state != streamgate.CommitStateTerminalCommitted {
|
|
t.Errorf("first commit state: want terminal_committed, got %v", state)
|
|
}
|
|
if sink.committedCount != 1 {
|
|
t.Errorf("committedCount = %d, want 1", sink.committedCount)
|
|
}
|
|
|
|
// Verify Responses JSON framing: just the envelope.
|
|
jsonOutput, err := encodeResponsesJSONError(sink.committedTerminal)
|
|
if err != nil {
|
|
t.Fatalf("encodeResponsesJSONError: %v", err)
|
|
}
|
|
|
|
expectedEnvelope := formatErrorEnvelope(sink.committedTerminal)
|
|
if jsonOutput != expectedEnvelope {
|
|
t.Errorf("Responses JSON output:\n got: %q\nwant: %q", jsonOutput, expectedEnvelope)
|
|
}
|
|
|
|
// Responses JSON must NOT contain SSE framing markers.
|
|
if strings.Contains(jsonOutput, "data:") {
|
|
t.Errorf("Responses JSON must not contain SSE 'data:' marker: %q", jsonOutput)
|
|
}
|
|
if strings.Contains(jsonOutput, "[DONE]") {
|
|
t.Errorf("Responses JSON must not contain SSE '[DONE]' marker: %q", jsonOutput)
|
|
}
|
|
|
|
// Decode the envelope and verify exact key set.
|
|
var decoded map[string]interface{}
|
|
if err := json.Unmarshal([]byte(jsonOutput), &decoded); err != nil {
|
|
t.Fatalf("outputBody JSON decode failed: %v; body=%q", err, jsonOutput)
|
|
}
|
|
|
|
if _, ok := decoded["error"]; !ok {
|
|
t.Errorf("decoded top-level keys: missing 'error'; got %+v", decoded)
|
|
}
|
|
topKeys := make([]string, 0, len(decoded))
|
|
for k := range decoded {
|
|
topKeys = append(topKeys, k)
|
|
}
|
|
if len(topKeys) != 1 || topKeys[0] != "error" {
|
|
t.Errorf("decoded top-level keys: want ['error'], got %v", topKeys)
|
|
}
|
|
|
|
errObj, ok := decoded["error"].(map[string]interface{})
|
|
if !ok {
|
|
t.Fatalf("decoded['error'] is not object; got %T", decoded["error"])
|
|
}
|
|
errKeys := make([]string, 0, len(errObj))
|
|
for k := range errObj {
|
|
errKeys = append(errKeys, k)
|
|
}
|
|
if len(errKeys) != 2 {
|
|
t.Errorf("decoded['error'] keys: want 2, got %d (%v)", len(errKeys), errKeys)
|
|
}
|
|
if _, ok := errObj["type"]; !ok {
|
|
t.Error("decoded['error'] missing 'type'")
|
|
}
|
|
if _, ok := errObj["message"]; !ok {
|
|
t.Error("decoded['error'] missing 'message'")
|
|
}
|
|
if errObj["type"] != "rate_limit_error" {
|
|
t.Errorf("decoded['error'].type = %v, want rate_limit_error", errObj["type"])
|
|
}
|
|
if errObj["message"] != "rate_limit_exceeded" {
|
|
t.Errorf("decoded['error'].message = %v, want 'rate_limit_exceeded'", errObj["message"])
|
|
}
|
|
|
|
// Forbidden internal fields must not appear anywhere in the JSON output.
|
|
forbiddenFields := []string{"causes", "stack", "trace", "provider", "internal", "raw", "repetition_loop", "filter_blocked", "validation_failed", "cache_miss", "stage", "consumer"}
|
|
for _, f := range forbiddenFields {
|
|
if strings.Contains(jsonOutput, f) {
|
|
t.Errorf("forbidden field '%s' found in Responses JSON output: %s", f, jsonOutput)
|
|
}
|
|
}
|
|
|
|
// Verify the committed terminal reflects the bounded cause chain.
|
|
committed := sink.committedTerminal
|
|
if !committed.Error() {
|
|
t.Error("committed terminal should be error type")
|
|
}
|
|
committedCauses := committed.FailureCauses()
|
|
if committedCauses.Len() != 4 {
|
|
t.Errorf("committed causes Len() = %d, want 4", committedCauses.Len())
|
|
}
|
|
})
|
|
|
|
// --- Second commit must be rejected without changing output ---
|
|
t.Run("second_commit_rejected", func(t *testing.T) {
|
|
secondSink := &fakeHostReleaseSink{}
|
|
secondSink.state = streamgate.CommitStateStreamOpen
|
|
|
|
// First commit should succeed.
|
|
_, err := secondSink.CommitTerminal(ctx, tr)
|
|
if err != nil {
|
|
t.Fatalf("first CommitTerminal in second-commit check should succeed: %v", err)
|
|
}
|
|
secondSinkCommittedCount := secondSink.committedCount
|
|
secondSinkOutputBody := secondSink.outputBody
|
|
secondSinkState := secondSink.state
|
|
|
|
// Second commit must be rejected.
|
|
_, err = secondSink.CommitTerminal(ctx, tr)
|
|
if err == nil {
|
|
t.Error("second CommitTerminal should have been rejected")
|
|
}
|
|
|
|
if secondSink.committedCount != secondSinkCommittedCount {
|
|
t.Errorf("committedCount after second: %d, want %d", secondSink.committedCount, secondSinkCommittedCount)
|
|
}
|
|
|
|
if secondSink.outputBody != secondSinkOutputBody {
|
|
t.Errorf("outputBody changed after second commit: %q != %q", secondSink.outputBody, secondSinkOutputBody)
|
|
}
|
|
|
|
if secondSink.state != secondSinkState {
|
|
t.Errorf("state changed after second commit: %q != %q", secondSink.state, secondSinkState)
|
|
}
|
|
})
|
|
}
|