diff --git a/runtime/argschema/argschema.go b/runtime/argschema/argschema.go new file mode 100644 index 0000000..f9c9892 --- /dev/null +++ b/runtime/argschema/argschema.go @@ -0,0 +1,85 @@ +// Package argschema checks tool-call arguments against the small JSON-schema +// subset that tool declarations use. It has no dependencies beyond the +// standard library so both the tool registry and the LLM client can share it. +package argschema + +import ( + "encoding/json" + "fmt" + "strings" +) + +// Validate reports whether args satisfy schema: the top-level type is object, +// required keys are present and not blank strings, declared properties have +// their primitive type, and unknown keys are rejected when properties are +// declared. +func Validate(schema, args json.RawMessage) error { + var spec struct { + Type string `json:"type"` + Required []string `json:"required"` + Properties map[string]json.RawMessage `json:"properties"` + } + if len(schema) > 0 { + if err := json.Unmarshal(schema, &spec); err != nil { + return fmt.Errorf("invalid tool schema") + } + } + var obj map[string]any + if err := json.Unmarshal(args, &obj); err != nil { + return fmt.Errorf("arguments must be a JSON object") + } + if spec.Type != "" && spec.Type != "object" { + return fmt.Errorf("unsupported schema type %s", spec.Type) + } + for _, req := range spec.Required { + v, ok := obj[req] + if !ok || v == nil { + return fmt.Errorf("missing required argument %s", req) + } + if s, ok := v.(string); ok && strings.TrimSpace(s) == "" { + return fmt.Errorf("missing required argument %s", req) + } + } + for key, raw := range spec.Properties { + v, ok := obj[key] + if !ok { + continue + } + var prop struct { + Type string `json:"type"` + } + if err := json.Unmarshal(raw, &prop); err != nil || prop.Type == "" { + continue + } + if !matchesType(prop.Type, v) { + return fmt.Errorf("argument %s must be %s", key, prop.Type) + } + } + if len(spec.Properties) > 0 { + for key := range obj { + if _, ok := spec.Properties[key]; !ok { + return fmt.Errorf("unknown argument %s", key) + } + } + } + return nil +} + +func matchesType(kind string, v any) bool { + switch kind { + case "string": + _, ok := v.(string) + return ok + case "number": + _, ok := v.(float64) + return ok + case "integer": + f, ok := v.(float64) + return ok && f == float64(int64(f)) + case "boolean": + _, ok := v.(bool) + return ok + default: + return true + } +} diff --git a/runtime/argschema/argschema_test.go b/runtime/argschema/argschema_test.go new file mode 100644 index 0000000..12614fc --- /dev/null +++ b/runtime/argschema/argschema_test.go @@ -0,0 +1,44 @@ +package argschema + +import ( + "encoding/json" + "testing" +) + +func TestValidate(t *testing.T) { + schema := json.RawMessage(`{"type":"object","required":["path"],"properties":{"path":{"type":"string"},"limit":{"type":"integer"},"ratio":{"type":"number"},"all":{"type":"boolean"},"any":{}}}`) + tests := []struct { + name string + schema json.RawMessage + args string + want string + }{ + {"valid", schema, `{"path":"a.txt","limit":3,"ratio":0.5,"all":true,"any":[1]}`, ""}, + {"no-schema-accepts-object", nil, `{"x":1}`, ""}, + {"no-properties-accepts-unknown", json.RawMessage(`{"type":"object"}`), `{"x":1}`, ""}, + {"invalid-schema", json.RawMessage(`[`), `{}`, "invalid tool schema"}, + {"array-arguments", schema, `[1]`, "arguments must be a JSON object"}, + {"broken-arguments", schema, `{"path":`, "arguments must be a JSON object"}, + {"non-object-schema", json.RawMessage(`{"type":"array"}`), `{}`, "unsupported schema type array"}, + {"missing-required", schema, `{}`, "missing required argument path"}, + {"null-required", schema, `{"path":null}`, "missing required argument path"}, + {"blank-required", schema, `{"path":" "}`, "missing required argument path"}, + {"wrong-string", schema, `{"path":7}`, "argument path must be string"}, + {"fractional-integer", schema, `{"path":"a","limit":1.5}`, "argument limit must be integer"}, + {"wrong-number", schema, `{"path":"a","ratio":"x"}`, "argument ratio must be number"}, + {"wrong-boolean", schema, `{"path":"a","all":1}`, "argument all must be boolean"}, + {"unknown-key", schema, `{"path":"a","extra":1}`, "unknown argument extra"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := Validate(tc.schema, json.RawMessage(tc.args)) + got := "" + if err != nil { + got = err.Error() + } + if got != tc.want { + t.Fatalf("Validate(%s, %s) = %q, want %q", tc.schema, tc.args, got, tc.want) + } + }) + } +} diff --git a/runtime/llm/client.go b/runtime/llm/client.go index 766208b..295a5a0 100644 --- a/runtime/llm/client.go +++ b/runtime/llm/client.go @@ -26,6 +26,9 @@ type Client struct { // SSE 的 reasoning_content 捕获照常生效)。对不认识该字段的服务器 // 属未知字段,多数按忽略处理;宿主按模型名自行决定是否置位。 ReasoningSplit bool + // RouteContract optionally releases tool calls from an openai-chat stream + // that ended without [DONE]; nil or the zero value is off. + RouteContract *RouteContractEOF } // Stream implements StreamFunc. Transport and provider failures are encoded as @@ -127,12 +130,36 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool return } observed := newBodyObserver(resp) - for ev := range ParseSSE(ctx, observed) { + contract := c.newRouteContract(model, base, req, resp, tools) + var facts sseFacts + var hooks sseHooks + if contract != nil { + hooks.facts = &facts + hooks.decideEOF = func(success, failure Event) Event { + check := contract.check(observed.snapshot(), failure.TerminalObservation, facts, success.Message.ToolCalls) + if check != "" { + failure.TerminalObservation.ContractCheck = check + return failure + } + success.CompletionBasis = CompletionRouteContractEOF + success.TerminalObservation.CompletionBasis = CompletionRouteContractEOF + return success + } + } + for ev := range parseSSE(ctx, observed, hooks) { if ev.Kind == KindMessageEnd || ev.Kind == KindError { ev.HTTPStatus = resp.StatusCode // ParseSSE sends its terminal event only after its last body read. - if ev.TerminalObservation != nil { - ev.TerminalObservation.BodyBoundary = observed.snapshot() + if obs := ev.TerminalObservation; obs != nil { + obs.BodyBoundary = observed.snapshot() + if ev.Kind == KindMessageEnd && ev.Err == nil && ev.StreamComplete { + ev.CompletionBasis = CompletionDoneMarker + obs.CompletionBasis = CompletionDoneMarker + } + if contract != nil && ev.Err != nil && !ev.StreamComplete && + ev.Message.StopReason != StopAborted && obs.ContractCheck == "" { + obs.ContractCheck = contract.check(obs.BodyBoundary, obs, facts, ev.Message.ToolCalls) + } } } out <- ev diff --git a/runtime/llm/route_contract.go b/runtime/llm/route_contract.go new file mode 100644 index 0000000..db6f2c0 --- /dev/null +++ b/runtime/llm/route_contract.go @@ -0,0 +1,164 @@ +package llm + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/url" + "strings" + + "agent-native/runtime/argschema" +) + +// Completion bases recorded on a successful OpenAI-chat message_end. +const ( + CompletionDoneMarker = "done_marker" + CompletionRouteContractEOF = "route_contract_eof" +) + +// RouteContractEOF lets one exact OpenAI-chat route release tool calls from a +// stream that ended without [DONE], when the HTTP body boundary and every +// frame prove the calls complete. The zero value is off. IdentitySHA256 must +// equal RouteContractIdentity(Model, BaseURL), and BaseURL must already be in +// NormalizeRouteBaseURL form. +type RouteContractEOF struct { + Model string + BaseURL string + IdentitySHA256 string +} + +// RouteContractIdentity is the lowercase hex sha256 of +// "openai-chat\n" + model + "\n" + baseURL. +func RouteContractIdentity(model, baseURL string) string { + sum := sha256.Sum256([]byte("openai-chat\n" + model + "\n" + baseURL)) + return hex.EncodeToString(sum[:]) +} + +// NormalizeRouteBaseURL returns the https base URL with lowercase scheme and +// host and no trailing '/'. Any other scheme, userinfo, query or fragment is +// not a match. +func NormalizeRouteBaseURL(raw string) (string, bool) { + if strings.ContainsAny(raw, "?#") { + return "", false + } + u, err := url.Parse(raw) + if err != nil || !strings.EqualFold(u.Scheme, "https") || u.Opaque != "" || + u.User != nil || u.Host == "" || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" { + return "", false + } + return "https://" + strings.ToLower(u.Host) + strings.TrimRight(u.EscapedPath(), "/"), true +} + +func (rc *RouteContractEOF) off() bool { + return rc == nil || *rc == RouteContractEOF{} +} + +// admits is condition C1 apart from the redirect check. +func (rc *RouteContractEOF) admits(protocol Protocol, model, base string) bool { + if protocol != "" && protocol != ProtocolOpenAIChat { + return false + } + if model != rc.Model { + return false + } + normalized, ok := NormalizeRouteBaseURL(base) + if !ok || normalized != rc.BaseURL { + return false + } + return RouteContractIdentity(rc.Model, rc.BaseURL) == rc.IdentitySHA256 +} + +// routeContract evaluates a configured RouteContractEOF for one response. +type routeContract struct { + active bool + tools []ToolDecl +} + +// newRouteContract returns nil when the client has no contract configured. +func (c *Client) newRouteContract(model, base string, req *http.Request, resp *http.Response, tools []ToolDecl) *routeContract { + if c.RouteContract.off() { + return nil + } + return &routeContract{ + active: c.RouteContract.admits(c.Protocol, model, base) && sameTarget(req, resp), + tools: tools, + } +} + +// sameTarget rejects any redirect that changed scheme, host, port or path. +func sameTarget(req *http.Request, resp *http.Response) bool { + if req == nil || req.URL == nil || resp == nil || resp.Request == nil || resp.Request.URL == nil { + return false + } + sent, final := req.URL, resp.Request.URL + return strings.EqualFold(sent.Scheme, final.Scheme) && + strings.EqualFold(sent.Hostname(), final.Hostname()) && + sent.Port() == final.Port() && + sent.EscapedPath() == final.EscapedPath() +} + +// check returns the first failing condition id, or "" when all hold. +func (rc *routeContract) check(boundary *BodyBoundaryObservation, obs *TerminalObservation, facts sseFacts, calls []ToolCall) string { + switch { + case !rc.active: + return "C1" + case !boundaryProven(boundary): + return "C2" + case obs == nil || obs.ReadOutcome != "clean_eof": + return "C3" + case facts.finishChunks != 1 || facts.finishReason != "tool_calls" || facts.multiChoice: + return "C4" + case !obs.UsagePresent: + return "C5" + case facts.fragmentAfterFinish: + return "C6" + case !callsComplete(calls): + return "C7" + case !callsDeclared(calls, rc.tools): + return "C8" + case obs.TailEventKind != "json_frame" || !obs.LastFrameJSONComplete: + return "C9" + } + return "" +} + +func boundaryProven(b *BodyBoundaryObservation) bool { + if b == nil { + return false + } + return (b.FramingClass == "http2_stream" && b.EOFSeen) || + (b.FramingClass == "known_length" && b.LengthRelation == "match") +} + +func callsComplete(calls []ToolCall) bool { + if len(calls) == 0 { + return false + } + for _, call := range calls { + var args map[string]json.RawMessage + if call.ID == "" || call.Name == "" { + return false + } + if err := json.Unmarshal([]byte(call.Arguments), &args); err != nil || args == nil { + return false + } + } + return true +} + +func callsDeclared(calls []ToolCall, tools []ToolDecl) bool { + for _, call := range calls { + idx := -1 + for i, decl := range tools { + if decl.Name == call.Name { + idx = i + break + } + } + if idx < 0 || argschema.Validate(tools[idx].Parameters, json.RawMessage(call.Arguments)) != nil { + return false + } + } + return true +} diff --git a/runtime/llm/route_contract_test.go b/runtime/llm/route_contract_test.go new file mode 100644 index 0000000..87af915 --- /dev/null +++ b/runtime/llm/route_contract_test.go @@ -0,0 +1,630 @@ +package llm + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strconv" + "strings" + "testing" +) + +const ( + rcModel = "fixture-model" + rcEOFReason = "tool call stream ended before [DONE]" +) + +var rcTools = []ToolDecl{ + {Name: "read_PRIVATE_I1", Parameters: json.RawMessage(`{"type":"object","required":["path"],"properties":{"path":{"type":"string"}}}`)}, + {Name: "list_PRIVATE_I1", Parameters: json.RawMessage(`{"type":"object","properties":{"dir":{"type":"string"}}}`)}, +} + +func rcFrame(v any) string { + raw, err := json.Marshal(v) + if err != nil { + panic(err) + } + return "data: " + string(raw) + "\n\n" +} + +func rcChunk(choices ...map[string]any) string { + return rcFrame(map[string]any{"choices": choices}) +} + +func rcFrag(index int, id, name, args string) map[string]any { + fn := map[string]any{} + if name != "" { + fn["name"] = name + } + if args != "" { + fn["arguments"] = args + } + frag := map[string]any{"index": index, "function": fn} + if id != "" { + frag["id"] = id + } + return frag +} + +func rcToolChoice(finish string, frags ...map[string]any) map[string]any { + choice := map[string]any{"delta": map[string]any{"tool_calls": frags}} + if finish != "" { + choice["finish_reason"] = finish + } + return choice +} + +func rcFinishChunk(reason string) string { + return rcChunk(map[string]any{"delta": map[string]any{}, "finish_reason": reason}) +} + +var ( + rcText = rcChunk(map[string]any{"delta": map[string]any{"content": "PRIVATE_TEXT_I1", "reasoning_content": "PRIVATE_THINK_I1"}}) + rcCall1 = rcChunk(rcToolChoice("", rcFrag(0, "call_PRIVATE_I1", "read_PRIVATE_I1", `{"path":`))) + rcCall2 = rcChunk(rcToolChoice("", rcFrag(0, "", "", `"PRIVATE_I1.txt"}`))) + rcFinish = rcFinishChunk("tool_calls") + rcUsage = rcFrame(map[string]any{"choices": []any{}, "usage": map[string]any{"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7}}) + rcDone = "data: [DONE]\n\n" + rcStream = rcText + rcCall1 + rcCall2 + rcFinish + rcUsage + rcCall = ToolCall{ID: "call_PRIVATE_I1", Name: "read_PRIVATE_I1", Arguments: `{"path":"PRIVATE_I1.txt"}`} + + rcErrorFrame = rcFrame(map[string]any{"error": map[string]any{"message": "PRIVATE_PROVIDER_I1"}}) +) + +// rcSingleCall streams one complete call in one fragment with the given parts. +func rcSingleCall(id, name, args string) string { + return rcText + rcChunk(rcToolChoice("", rcFrag(0, id, name, args))) + rcFinish + rcUsage +} + +func rcWrite(stream string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, stream) + w.(http.Flusher).Flush() + } +} + +func rcHTTP2(t *testing.T, h http.Handler) *httptest.Server { + t.Helper() + server := httptest.NewUnstartedServer(h) + server.EnableHTTP2 = true + server.StartTLS() + t.Cleanup(server.Close) + return server +} + +func rcTLS1(t *testing.T, h http.Handler) *httptest.Server { + t.Helper() + server := httptest.NewTLSServer(h) + t.Cleanup(server.Close) + return server +} + +func rcPlain(t *testing.T, h http.Handler) *httptest.Server { + t.Helper() + server := httptest.NewServer(h) + t.Cleanup(server.Close) + return server +} + +// rcHTTPClient trusts server like server.Client() but owns its connection +// pool, so each run starts on a fresh connection and HTTP/2 stream ids match. +func rcHTTPClient(t *testing.T, server *httptest.Server) *http.Client { + t.Helper() + base := server.Client().Transport.(*http.Transport) + transport := &http.Transport{TLSClientConfig: base.TLSClientConfig.Clone(), ForceAttemptHTTP2: true} + t.Cleanup(transport.CloseIdleConnections) + return &http.Client{Transport: transport} +} + +func rcClientFor(t *testing.T, server *httptest.Server) *Client { + return &Client{BaseURL: server.URL, APIKey: "fixture", HTTPClient: rcHTTPClient(t, server)} +} + +func rcContract(model, base string) *RouteContractEOF { + return &RouteContractEOF{Model: model, BaseURL: base, IdentitySHA256: RouteContractIdentity(model, base)} +} + +func rcFixtureClient(resp func() *http.Response) *Client { + return &Client{BaseURL: "https://fixture.invalid", APIKey: "fixture", HTTPClient: &http.Client{ + Transport: terminalRoundTripFunc(func(r *http.Request) (*http.Response, error) { + res := resp() + res.Request = r + return res, nil + }), + }} +} + +type rcSetup struct { + client *Client + contract *RouteContractEOF + opts StreamOptions + tools []ToolDecl +} + +type rcResult struct { + end Event + releasedMessages int + releasedCalls int +} + +func rcRun(t *testing.T, ctx context.Context, base *Client, contract *RouteContractEOF, tools []ToolDecl, opts StreamOptions, onEvent func(Event)) rcResult { + t.Helper() + client := *base + client.RouteContract = contract + var res rcResult + terminals := 0 + for ev := range client.Stream(ctx, rcModel, []Message{{Role: RoleUser, Content: "hello"}}, tools, opts) { + if onEvent != nil { + onEvent(ev) + } + if ev.Kind != KindMessageEnd && ev.Kind != KindError { + continue + } + terminals++ + res.end = ev + if ev.Kind == KindMessageEnd && ev.Err == nil && len(ev.Message.ToolCalls) > 0 { + res.releasedMessages++ + res.releasedCalls += len(ev.Message.ToolCalls) + } + } + if terminals != 1 { + t.Fatalf("terminal events = %d, want 1", terminals) + } + rcRequireSafeLabels(t, res.end) + return res +} + +// rcRequireSafeLabels is the sentinel scan: the new labels hold only their +// fixed enums and the observation never carries content, tool identity, +// arguments or error text. +func rcRequireSafeLabels(t *testing.T, end Event) { + t.Helper() + obs := end.TerminalObservation + if obs == nil { + t.Fatalf("terminal without observation: %#v", end) + } + switch end.CompletionBasis { + case "", CompletionDoneMarker, CompletionRouteContractEOF: + default: + t.Fatalf("completion basis outside enum: %q", end.CompletionBasis) + } + if obs.CompletionBasis != end.CompletionBasis { + t.Fatalf("observation basis %q != event basis %q", obs.CompletionBasis, end.CompletionBasis) + } + if c := obs.ContractCheck; c != "" && (len(c) != 2 || c[0] != 'C' || c[1] < '1' || c[1] > '9') { + t.Fatalf("contract check outside enum: %q", c) + } + raw, err := json.Marshal(obs) + if err != nil { + t.Fatal(err) + } + for _, secret := range []string{"PRIVATE", "[DONE]", "tool call", "fixture-model"} { + if strings.Contains(string(raw), secret) { + t.Fatalf("observation leaked %q: %s", secret, raw) + } + } + for _, label := range []string{end.CompletionBasis, obs.CompletionBasis, obs.ContractCheck} { + if strings.Contains(label, "PRIVATE") || strings.Contains(label, " ") { + t.Fatalf("label carries content: %q", label) + } + } +} + +func rcStreamCase(server func(*testing.T, http.Handler) *httptest.Server, stream string) func(*testing.T) rcSetup { + return func(t *testing.T) rcSetup { + s := server(t, rcWrite(stream)) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + } +} + +func rcH2(stream string) func(*testing.T) rcSetup { return rcStreamCase(rcHTTP2, stream) } + +func rcRedirectCase(target func(t *testing.T) string) func(*testing.T) rcSetup { + return func(t *testing.T) rcSetup { + location := target(t) + mux := http.NewServeMux() + mux.HandleFunc("/chat/completions", func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, location, http.StatusTemporaryRedirect) + }) + mux.HandleFunc("/final/chat/completions", rcWrite(rcStream)) + s := rcHTTP2(t, mux) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + } +} + +type rcCase struct { + name string + setup func(*testing.T) rcSetup + // want is done_marker, route_contract_eof, off, or a check id C1..C9. + want string + // reason is the contract-off failure text; empty skips the check. + reason string + calls []ToolCall + noUsage bool +} + +func rcCases() []rcCase { + twoCalls := rcText + rcCall1 + rcCall2 + + rcChunk(rcToolChoice("", rcFrag(1, "call_PRIVATE_I1_b", "list_PRIVATE_I1", `{}`))) + rcFinish + rcUsage + return []rcCase{ + // [DONE] keeps today's path whether or not the contract is configured. + {name: "done/contract-on", setup: rcH2(rcStream + rcDone), want: CompletionDoneMarker}, + {name: "done/contract-c1-failing", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream + rcDone)(t) + s.contract = rcContract("other-model", s.client.BaseURL) + return s + }, want: CompletionDoneMarker}, + {name: "done/known-length", setup: rcKnownLength(rcStream + rcDone), want: CompletionDoneMarker}, + + // Released only when every condition holds. + {name: "release/http2", setup: rcH2(rcStream), want: CompletionRouteContractEOF}, + {name: "release/known-length-match", setup: rcKnownLength(rcStream), want: CompletionRouteContractEOF}, + {name: "release/usage-before-finish", setup: rcH2(rcText + rcUsage + rcCall1 + rcCall2 + rcFinish), want: CompletionRouteContractEOF}, + {name: "release/usage-after-finish", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinish + rcUsage), want: CompletionRouteContractEOF}, + {name: "release/fragment-in-finish-chunk", setup: rcH2(rcText + rcCall1 + + rcChunk(rcToolChoice("tool_calls", rcFrag(0, "", "", `"PRIVATE_I1.txt"}`))) + rcUsage), want: CompletionRouteContractEOF}, + {name: "release/two-calls", setup: rcH2(twoCalls), want: CompletionRouteContractEOF, + calls: []ToolCall{rcCall, {ID: "call_PRIVATE_I1_b", Name: "list_PRIVATE_I1", Arguments: `{}`}}}, + {name: "release/api-base-override-to-contract", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.opts.APIBase = s.client.BaseURL + "/" + s.client.BaseURL = "https://unused.invalid" + return s + }, want: CompletionRouteContractEOF}, + + // The zero value is off. + {name: "off/zero-contract", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.contract = &RouteContractEOF{} + return s + }, want: "off", reason: rcEOFReason}, + + // C1 route identity. + {name: "c1/model-mismatch", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.contract = rcContract("other-model", s.client.BaseURL) + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/base-http", setup: rcStreamCase(rcPlain, rcStream), want: "C1", reason: rcEOFReason}, + {name: "c1/base-userinfo", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.client.BaseURL = strings.Replace(s.client.BaseURL, "https://", "https://user@", 1) + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/base-query", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.client.BaseURL += "?x=1" + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/digest-mismatch", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.contract.IdentitySHA256 = strings.Repeat("0", 64) + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/digest-uppercase", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.contract.IdentitySHA256 = strings.ToUpper(s.contract.IdentitySHA256) + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/protocol-not-openai-chat", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.client.Protocol = ProtocolGoogle + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/api-base-override-elsewhere", setup: func(t *testing.T) rcSetup { + other := rcHTTP2(t, rcWrite(rcStream)) + s := rcH2(rcStream)(t) + s.opts.APIBase = other.URL + s.client.HTTPClient = rcHTTPClient(t, other) + return s + }, want: "C1", reason: rcEOFReason}, + {name: "c1/redirect-other-host", setup: rcRedirectCase(func(t *testing.T) string { + return rcHTTP2(t, rcWrite(rcStream)).URL + "/chat/completions" + }), want: "C1", reason: rcEOFReason}, + {name: "c1/redirect-same-host-other-path", setup: rcRedirectCase(func(t *testing.T) string { + return "/final/chat/completions" + }), want: "C1", reason: rcEOFReason}, + {name: "c1/redirect-https-downgrade", setup: rcRedirectCase(func(t *testing.T) string { + return rcPlain(t, rcWrite(rcStream)).URL + "/chat/completions" + }), want: "C1", reason: rcEOFReason}, + + // C2 body boundary. + {name: "c2/known-length-mismatch", setup: func(t *testing.T) rcSetup { + client := rcFixtureClient(func() *http.Response { + return &http.Response{StatusCode: http.StatusOK, ProtoMajor: 1, ProtoMinor: 1, + ContentLength: int64(len(rcStream) + 64), Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(rcStream))} + }) + return rcSetup{client: client, contract: rcContract(rcModel, client.BaseURL)} + }, want: "C2", reason: rcEOFReason}, + {name: "c2/unknown-framing", setup: func(t *testing.T) rcSetup { + client := rcFixtureClient(func() *http.Response { + return &http.Response{StatusCode: http.StatusOK, ContentLength: -1, Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(rcStream))} + }) + return rcSetup{client: client, contract: rcContract(rcModel, client.BaseURL)} + }, want: "C2", reason: rcEOFReason}, + {name: "c2/chunked", setup: rcStreamCase(rcTLS1, rcStream), want: "C2", reason: rcEOFReason}, + {name: "c2/close-delimited", setup: func(t *testing.T) rcSetup { + s := rcTLS1(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Transfer-Encoding", "identity") + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, rcStream) + w.(http.Flusher).Flush() + })) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + }, want: "C2", reason: rcEOFReason}, + // A reset or dropped stream never reaches body EOF, so the boundary + // condition fails before the read-outcome condition is evaluated. + {name: "c2/http2-rst-stream", setup: func(t *testing.T) rcSetup { + s := rcHTTP2(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rcWrite(rcStream)(w, r) + panic(http.ErrAbortHandler) + })) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + }, want: "C2", reason: "stream read failed after reported usage"}, + {name: "c2/mid-stream-disconnect", setup: func(t *testing.T) rcSetup { + s := rcTLS1(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rcWrite(rcStream)(w, r) + conn, _, err := w.(http.Hijacker).Hijack() + if err == nil { + _ = conn.Close() + } + })) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + }, want: "C2", reason: "stream read failed after reported usage"}, + + // The parser stops at a bad frame; with a trailing delimiter it never + // reads to body EOF, so the boundary condition fails first. + {name: "c2/invalid-frame-before-eof", setup: rcH2(rcStream + "data: PRIVATE_not_json\n\n"), want: "C2", + reason: "malformed stream chunk after reported usage"}, + {name: "c2/provider-error-before-eof", setup: rcH2(rcStream + rcErrorFrame), want: "C2", + reason: "provider error after reported usage: PRIVATE_PROVIDER_I1"}, + + // C3 read outcome: the bad frame is the last bytes, so body EOF is seen. + {name: "c3/half-frame", setup: rcH2(rcStream + `data: {"choices":[{"delta":{"content":"PRIVATE_HALF`), want: "C3", + reason: "malformed stream chunk after reported usage"}, + {name: "c3/invalid-frame", setup: rcH2(rcStream + "data: PRIVATE_not_json"), want: "C3", + reason: "malformed stream chunk after reported usage"}, + {name: "c3/provider-error-after-usage", setup: rcH2(rcStream + strings.TrimRight(rcErrorFrame, "\n")), + want: "C3", reason: "provider error after reported usage: PRIVATE_PROVIDER_I1"}, + {name: "c3/provider-error-before-usage", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinish + strings.TrimRight(rcErrorFrame, "\n")), + want: "C3", reason: "PRIVATE_PROVIDER_I1", noUsage: true}, + + // C4 finish reason and choices. + {name: "c4/two-finish-chunks", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinish + rcFinish + rcUsage), want: "C4", reason: rcEOFReason}, + {name: "c4/finish-stop-with-tools", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinishChunk("stop") + rcUsage), want: "C4", reason: rcEOFReason}, + {name: "c4/finish-length-with-tools", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinishChunk("length") + rcUsage), want: "C4", reason: rcEOFReason}, + {name: "c4/multi-choice-chunk", setup: rcH2(rcText + + rcChunk(map[string]any{"delta": map[string]any{"content": "a"}}, map[string]any{"delta": map[string]any{"content": "b"}}) + + rcCall1 + rcCall2 + rcFinish + rcUsage), want: "C4", reason: rcEOFReason}, + {name: "c4/truncated-at-frame-boundary-without-finish", setup: rcH2(rcText + rcCall1 + rcCall2 + rcUsage), want: "C4", reason: rcEOFReason}, + + // C5 usage. + {name: "c5/missing-usage", setup: rcH2(rcText + rcCall1 + rcCall2 + rcFinish), want: "C5", reason: rcEOFReason, noUsage: true}, + + // C6 fragments after finish. + {name: "c6/fragment-after-finish-chunk", setup: rcH2(rcText + rcCall1 + rcFinish + rcCall2 + rcUsage), want: "C6", reason: rcEOFReason}, + + // C7 call identity and argument shape. + {name: "c7/missing-id", setup: rcH2(rcSingleCall("", "read_PRIVATE_I1", `{"path":"PRIVATE_I1.txt"}`)), want: "C7", reason: rcEOFReason}, + {name: "c7/missing-name", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "", `{"path":"PRIVATE_I1.txt"}`)), want: "C7", reason: rcEOFReason}, + {name: "c7/array-arguments", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `["PRIVATE_I1.txt"]`)), want: "C7", reason: rcEOFReason}, + {name: "c7/broken-arguments", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `{"path":"PRIVATE_I1.txt"`)), want: "C7", reason: rcEOFReason}, + {name: "c7/finish-without-call", setup: rcH2(rcText + rcFinish + rcUsage), want: "C7", reason: "tool call finish reason without a tool call"}, + + // C8 declared tools and argument schema. + {name: "c8/unknown-tool", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "write_PRIVATE_I1", `{"path":"PRIVATE_I1.txt"}`)), want: "C8", reason: rcEOFReason}, + {name: "c8/no-tools-declared", setup: func(t *testing.T) rcSetup { + s := rcH2(rcStream)(t) + s.tools = []ToolDecl{} + return s + }, want: "C8", reason: rcEOFReason}, + {name: "c8/missing-required", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `{}`)), want: "C8", reason: rcEOFReason}, + {name: "c8/blank-required", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `{"path":" "}`)), want: "C8", reason: rcEOFReason}, + {name: "c8/wrong-property-type", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `{"path":7}`)), want: "C8", reason: rcEOFReason}, + {name: "c8/unknown-argument", setup: rcH2(rcSingleCall("call_PRIVATE_I1", "read_PRIVATE_I1", `{"path":"a","PRIVATE":1}`)), want: "C8", reason: rcEOFReason}, + + // C9 tail event. + {name: "c9/comment-tail", setup: rcH2(rcStream + ": PRIVATE_keepalive\n\n"), want: "C9", reason: rcEOFReason}, + } +} + +func rcKnownLength(stream string) func(*testing.T) rcSetup { + return func(t *testing.T) rcSetup { + s := rcTLS1(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Content-Length", strconv.Itoa(len(stream))) + _, _ = io.WriteString(w, stream) + })) + return rcSetup{client: rcClientFor(t, s), contract: rcContract(rcModel, s.URL)} + } +} + +func TestRouteContractEOFMatrix(t *testing.T) { + for _, tc := range rcCases() { + t.Run(tc.name, func(t *testing.T) { + s := tc.setup(t) + tools := s.tools + if tools == nil { + tools = rcTools + } + wantCalls := tc.calls + if wantCalls == nil { + wantCalls = []ToolCall{rcCall} + } + wantUsage := 7 + if tc.noUsage { + wantUsage = 0 + } + off := rcRun(t, context.Background(), s.client, nil, tools, s.opts, nil) + on := rcRun(t, context.Background(), s.client, s.contract, tools, s.opts, nil) + for _, r := range []rcResult{off, on} { + if r.end.Usage.TotalTokens != wantUsage { + t.Fatalf("usage = %+v, want total %d", r.end.Usage, wantUsage) + } + } + if off.end.TerminalObservation.ContractCheck != "" { + t.Fatalf("contract-off stream carried a contract check: %+v", off.end.TerminalObservation) + } + if tc.reason != "" && off.end.ErrorMessage != tc.reason { + t.Fatalf("contract-off failure = %q, want %q", off.end.ErrorMessage, tc.reason) + } + + switch tc.want { + case CompletionDoneMarker: + if !reflect.DeepEqual(on.end, off.end) { + t.Fatalf("contract changed a [DONE] terminal:\n on=%#v\noff=%#v", on.end, off.end) + } + if on.end.Err != nil || !on.end.StreamComplete || on.end.CompletionBasis != CompletionDoneMarker || + on.end.TerminalObservation.ContractCheck != "" || on.end.TerminalObservation.ReadOutcome != "done_marker" { + t.Fatalf("[DONE] terminal not marked done_marker: %#v", on.end) + } + rcRequireReleased(t, on, wantCalls) + case CompletionRouteContractEOF: + rcRequireRejected(t, off) + if off.end.CompletionBasis != "" || off.end.ErrorMessage != rcEOFReason { + t.Fatalf("contract-off stream was not rejected unchanged: %#v", off.end) + } + end := on.end + if end.Err != nil || end.StreamComplete || end.CompletionBasis != CompletionRouteContractEOF || + end.TerminalObservation.ContractCheck != "" || end.TerminalObservation.ReadOutcome != "clean_eof" || + end.Message.StopReason != StopToolCalls || end.FinishReason != "tool_calls" || end.HTTPStatus != http.StatusOK { + t.Fatalf("contract stream was not released: %#v obs=%+v", end, end.TerminalObservation) + } + if end.Message.Content != off.end.Message.Content || !reflect.DeepEqual(end.Message.Thinking, off.end.Message.Thinking) { + t.Fatalf("released message changed content: %#v vs %#v", end.Message, off.end.Message) + } + rcRequireReleased(t, on, wantCalls) + case "off": + if !reflect.DeepEqual(on.end, off.end) { + t.Fatalf("zero contract changed the terminal:\n on=%#v\noff=%#v", on.end, off.end) + } + rcRequireRejected(t, on) + default: + rcRequireRejected(t, off) + rcRequireRejected(t, on) + if got := on.end.TerminalObservation.ContractCheck; got != tc.want { + t.Fatalf("contract check = %q, want %q (obs=%+v)", got, tc.want, on.end.TerminalObservation) + } + stripped := on.end + obs := *on.end.TerminalObservation + obs.ContractCheck = "" + stripped.TerminalObservation = &obs + if !reflect.DeepEqual(stripped, off.end) { + t.Fatalf("rejection differs from contract-off beyond contractCheck:\n on=%#v %+v\noff=%#v %+v", + stripped, stripped.TerminalObservation, off.end, off.end.TerminalObservation) + } + } + }) + } +} + +func rcRequireReleased(t *testing.T, r rcResult, calls []ToolCall) { + t.Helper() + if r.releasedMessages != 1 || r.releasedCalls != len(calls) || !reflect.DeepEqual(r.end.Message.ToolCalls, calls) { + t.Fatalf("released %d messages / %d calls %#v, want 1 / %#v", r.releasedMessages, r.releasedCalls, r.end.Message.ToolCalls, calls) + } +} + +func rcRequireRejected(t *testing.T, r rcResult) { + t.Helper() + end := r.end + if r.releasedMessages != 0 || r.releasedCalls != 0 || len(end.Message.ToolCalls) != 0 || end.Err == nil || + end.StreamComplete || end.CompletionBasis != "" || end.TerminalObservation.CompletionBasis != "" { + t.Fatalf("stream was released: %#v", end) + } +} + +func TestRouteContractEOFCancelDuringReadAborts(t *testing.T) { + server := rcHTTP2(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rcWrite(rcText+rcCall1+rcCall2+rcFinish+rcUsage)(w, r) + <-r.Context().Done() + })) + client := rcClientFor(t, server) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + res := rcRun(t, ctx, client, rcContract(rcModel, server.URL), rcTools, StreamOptions{}, func(ev Event) { + if ev.Kind == KindToolCallDelta { + cancel() + } + }) + end := res.end + if end.Kind != KindError || end.Message.StopReason != StopAborted || end.TerminalObservation.ReadOutcome != "cancelled" || + end.TerminalObservation.ContractCheck != "" || end.CompletionBasis != "" || res.releasedMessages != 0 || + res.releasedCalls != 0 || len(end.Message.ToolCalls) != 0 { + t.Fatalf("cancelled contract stream was not aborted: %#v obs=%+v", end, end.TerminalObservation) + } +} + +func TestParseSSEUnchangedWithoutDone(t *testing.T) { + var end Event + for ev := range ParseSSE(context.Background(), strings.NewReader(rcStream)) { + end = ev + } + if end.Kind != KindMessageEnd || end.ErrorMessage != rcEOFReason || len(end.Message.ToolCalls) != 0 || + end.StreamComplete || end.CompletionBasis != "" || end.Usage.TotalTokens != 7 || + end.TerminalObservation == nil || end.TerminalObservation.CompletionBasis != "" || + end.TerminalObservation.ContractCheck != "" || end.TerminalObservation.BodyBoundary != nil { + t.Fatalf("ParseSSE changed without [DONE]: %#v", end) + } + raw, err := json.Marshal(end.TerminalObservation) + if err != nil || strings.Contains(string(raw), "completionBasis") || strings.Contains(string(raw), "contractCheck") { + t.Fatalf("ParseSSE observation JSON changed: %s err=%v", raw, err) + } + var done Event + for ev := range ParseSSE(context.Background(), strings.NewReader(rcStream+rcDone)) { + done = ev + } + if done.Err != nil || !done.StreamComplete || done.CompletionBasis != "" || done.TerminalObservation.CompletionBasis != "" || + !reflect.DeepEqual(done.Message.ToolCalls, []ToolCall{rcCall}) { + t.Fatalf("ParseSSE changed with [DONE]: %#v", done) + } +} + +func TestRouteContractIdentity(t *testing.T) { + sum := sha256.Sum256([]byte("openai-chat\nm\nhttps://api.example.com/v1")) + if got, want := RouteContractIdentity("m", "https://api.example.com/v1"), hex.EncodeToString(sum[:]); got != want { + t.Fatalf("identity = %s, want %s", got, want) + } + if got := RouteContractIdentity("m", "https://api.example.com/v1"); got != strings.ToLower(got) || len(got) != 64 { + t.Fatalf("identity is not lowercase hex: %s", got) + } +} + +func TestNormalizeRouteBaseURL(t *testing.T) { + tests := []struct { + raw, want string + ok bool + }{ + {"https://api.example.com/v1", "https://api.example.com/v1", true}, + {"HTTPS://API.Example.COM/v1/", "https://api.example.com/v1", true}, + {"https://api.example.com:8443/v1//", "https://api.example.com:8443/v1", true}, + {"https://api.example.com", "https://api.example.com", true}, + {"https://api.example.com/V1", "https://api.example.com/V1", true}, + {"http://api.example.com/v1", "", false}, + {"https://user@api.example.com/v1", "", false}, + {"https://user:pw@api.example.com/v1", "", false}, + {"https://api.example.com/v1?x=1", "", false}, + {"https://api.example.com/v1?", "", false}, + {"https://api.example.com/v1#frag", "", false}, + {"https://api.example.com/v1#", "", false}, + {"https:api.example.com", "", false}, + {"https:///v1", "", false}, + {"", "", false}, + {"://bad", "", false}, + } + for _, tc := range tests { + got, ok := NormalizeRouteBaseURL(tc.raw) + if got != tc.want || ok != tc.ok { + t.Fatalf("NormalizeRouteBaseURL(%q) = %q, %v; want %q, %v", tc.raw, got, ok, tc.want, tc.ok) + } + } +} diff --git a/runtime/llm/sse.go b/runtime/llm/sse.go index 7ae4c35..d278a37 100644 --- a/runtime/llm/sse.go +++ b/runtime/llm/sse.go @@ -46,18 +46,49 @@ const ( maxObservedTailEvents = 10000 ) +// sseFacts are structural stream facts the route contract needs. They are +// written by the parser goroutine before it sends the terminal event. +type sseFacts struct { + // finishChunks counts chunks with a non-empty finish_reason. + finishChunks int + finishReason string + // multiChoice is set when any chunk carried more than one choice. + multiChoice bool + // fragmentAfterFinish is set when a tool-call fragment arrived in a chunk + // after the one carrying finish_reason. + fragmentAfterFinish bool +} + +// sseHooks are internal parser options. The zero value is ParseSSE. +type sseHooks struct { + facts *sseFacts + // decideEOF, when set, receives a tool stream whose terminal would be + // "tool call stream ended before [DONE]": success is the message_end the + // stream would end with had [DONE] arrived, failure is the event emitted + // today. It returns the event to send. + decideEOF func(success, failure Event) Event +} + // ParseSSE reads an OpenAI-compatible server-sent event body and emits // deltas plus one terminal message_end or error event. func ParseSSE(ctx context.Context, r io.Reader) <-chan Event { + return parseSSE(ctx, r, sseHooks{}) +} + +func parseSSE(ctx context.Context, r io.Reader, hooks sseHooks) <-chan Event { out := make(chan Event) go func() { defer close(out) - emitParsed(ctx, r, out) + emitParsed(ctx, r, out, hooks) }() return out } -func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { +func emitParsed(ctx context.Context, r io.Reader, out chan<- Event, hooks sseHooks) { + facts := hooks.facts + if facts == nil { + facts = &sseFacts{} + } var content strings.Builder var reasoning strings.Builder calls := map[int]*ToolCall{} @@ -93,20 +124,39 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { TailEventKind: tailKind, TailEventCount: reader.tailCount, } } - sendTerminalFailure := func(why string) { + buildTerminalFailure := func(why string) Event { message := Message{Role: RoleAssistant, Content: content.String(), StopReason: StopError} if reasoning.Len() > 0 { message.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}} } failed := terminalStreamFailure(message, usage, finished, sawDone, why) failed.TerminalObservation = observation() + return failed + } + sendTerminal := func(ev Event) { select { case <-ctx.Done(): readOutcome = "cancelled" emitAbort(out, content.String(), ctx.Err(), observation()) - case out <- failed: + case out <- ev: } } + sendTerminalFailure := func(why string) { + sendTerminal(buildTerminalFailure(why)) + } + buildMessageEnd := func() Event { + msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)} + if reasoning.Len() > 0 { + msg.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}} + } + for _, idx := range order { + msg.ToolCalls = append(msg.ToolCalls, *calls[idx]) + } + if msg.StopReason == StopToolCalls && len(msg.ToolCalls) == 0 { + msg.StopReason = StopEnd + } + return Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone, TerminalObservation: observation()} + } for { data, ok := reader.next() if !ok { @@ -155,12 +205,28 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { usage.CacheReadTokens = chunk.Usage.Details.Cached } } + if len(chunk.Choices) > 1 { + facts.multiChoice = true + } + finishedBefore := facts.finishChunks > 0 + for _, c := range chunk.Choices { + if c.FinishReason != "" { + if facts.finishChunks < maxObservedTailEvents { + facts.finishChunks++ + } + break + } + } if len(chunk.Choices) == 0 { continue } choice := chunk.Choices[0] if choice.FinishReason != "" { finished = choice.FinishReason + facts.finishReason = finished + } + if finishedBefore && len(choice.Delta.ToolCalls) > 0 { + facts.fragmentAfterFinish = true } if choice.Delta.Content != "" { content.WriteString(choice.Delta.Content) @@ -279,26 +345,15 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { } } if problem != "" { + if !sawDone && hooks.decideEOF != nil { + sendTerminal(hooks.decideEOF(buildMessageEnd(), buildTerminalFailure(problem))) + return + } sendTerminalFailure(problem) return } } - msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)} - if reasoning.Len() > 0 { - msg.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}} - } - for _, idx := range order { - msg.ToolCalls = append(msg.ToolCalls, *calls[idx]) - } - if msg.StopReason == StopToolCalls && len(msg.ToolCalls) == 0 { - msg.StopReason = StopEnd - } - select { - case <-ctx.Done(): - readOutcome = "cancelled" - emitAbort(out, content.String(), ctx.Err(), observation()) - case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone, TerminalObservation: observation()}: - } + sendTerminal(buildMessageEnd()) } func terminalStreamFailure(message Message, usage Usage, reason string, complete bool, why string) Event { diff --git a/runtime/llm/types.go b/runtime/llm/types.go index 1499deb..4dd1b4d 100644 --- a/runtime/llm/types.go +++ b/runtime/llm/types.go @@ -117,6 +117,12 @@ type TerminalObservation struct { // BodyBoundary is set only by Client on a response it received; a bare // ParseSSE call leaves it nil. BodyBoundary *BodyBoundaryObservation `json:"bodyBoundary,omitempty"` + // CompletionBasis mirrors Event.CompletionBasis. + CompletionBasis string `json:"completionBasis,omitempty"` + // ContractCheck is the first route-contract condition, C1 through C9, that + // failed when a configured RouteContractEOF did not release a stream that + // ended without [DONE]. Empty when no contract evaluated the stream. + ContractCheck string `json:"contractCheck,omitempty"` } // BodyBoundaryObservation records how the client read one HTTP response body. @@ -152,8 +158,13 @@ type Event struct { // 空表示这条事件不携带结束原因。StopReason 的映射不受此字段影响。 FinishReason string // StreamComplete records whether the OpenAI-compatible SSE stream supplied - // its explicit [DONE] marker. Tool calls require this before execution. + // its explicit [DONE] marker. Tool calls require this before execution + // unless CompletionBasis is route_contract_eof. StreamComplete bool + // CompletionBasis is set by Client on a successful OpenAI-chat message_end: + // done_marker when [DONE] arrived, route_contract_eof when RouteContractEOF + // released tool calls without it. Empty otherwise. + CompletionBasis string // TerminalObservation is present only on a terminal/error event. It does // not change StreamComplete or the tool-execution safety gate. TerminalObservation *TerminalObservation diff --git a/runtime/tools/registry.go b/runtime/tools/registry.go index 563acae..e9ab4ba 100644 --- a/runtime/tools/registry.go +++ b/runtime/tools/registry.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "agent-native/runtime/argschema" "agent-native/runtime/checkpoint" "agent-native/runtime/filestate" ) @@ -286,74 +287,7 @@ func (r *Registry) audit(name string, blocked, isError bool, note string) { } func validateArgs(schema, args json.RawMessage) error { - var spec struct { - Type string `json:"type"` - Required []string `json:"required"` - Properties map[string]json.RawMessage `json:"properties"` - } - if len(schema) > 0 { - if err := json.Unmarshal(schema, &spec); err != nil { - return fmt.Errorf("invalid tool schema") - } - } - var obj map[string]any - if err := json.Unmarshal(args, &obj); err != nil { - return fmt.Errorf("arguments must be a JSON object") - } - if spec.Type != "" && spec.Type != "object" { - return fmt.Errorf("unsupported schema type %s", spec.Type) - } - for _, req := range spec.Required { - v, ok := obj[req] - if !ok || v == nil { - return fmt.Errorf("missing required argument %s", req) - } - if s, ok := v.(string); ok && strings.TrimSpace(s) == "" { - return fmt.Errorf("missing required argument %s", req) - } - } - for key, raw := range spec.Properties { - v, ok := obj[key] - if !ok { - continue - } - var prop struct { - Type string `json:"type"` - } - if err := json.Unmarshal(raw, &prop); err != nil || prop.Type == "" { - continue - } - if !matchesType(prop.Type, v) { - return fmt.Errorf("argument %s must be %s", key, prop.Type) - } - } - if len(spec.Properties) > 0 { - for key := range obj { - if _, ok := spec.Properties[key]; !ok { - return fmt.Errorf("unknown argument %s", key) - } - } - } - return nil -} - -func matchesType(kind string, v any) bool { - switch kind { - case "string": - _, ok := v.(string) - return ok - case "number": - _, ok := v.(float64) - return ok - case "integer": - f, ok := v.(float64) - return ok && f == float64(int64(f)) - case "boolean": - _, ok := v.(bool) - return ok - default: - return true - } + return argschema.Validate(schema, args) } // WorkspaceTools registers the document workspace tool set. diff --git a/runtime/tools/registry_test.go b/runtime/tools/registry_test.go new file mode 100644 index 0000000..7de03b7 --- /dev/null +++ b/runtime/tools/registry_test.go @@ -0,0 +1,18 @@ +package tools + +import ( + "encoding/json" + "testing" + + "agent-native/runtime/argschema" +) + +func TestValidateArgsDelegatesToArgschema(t *testing.T) { + schema := json.RawMessage(`{"type":"object","required":["path"],"properties":{"path":{"type":"string"}}}`) + for _, args := range []string{`{"path":"a"}`, `{}`, `{"path":" "}`, `{"path":1}`, `{"path":"a","x":1}`, `[1]`} { + got, want := validateArgs(schema, json.RawMessage(args)), argschema.Validate(schema, json.RawMessage(args)) + if (got == nil) != (want == nil) || (got != nil && got.Error() != want.Error()) { + t.Fatalf("validateArgs(%s) = %v, argschema.Validate = %v", args, got, want) + } + } +}