From 1ceafe62a2853c094dca6c44120ae82d73746c25 Mon Sep 17 00:00:00 2001 From: iliya Date: Sun, 6 Sep 2026 06:28:04 +0000 Subject: [PATCH] fix(authoring): bound compatibility scan work and drain concurrent results --- .../internal/authoring/commands/slice_test.go | 116 ++++++++++---- .../conformance/root_compat_test.go | 146 ++++++++++++++++++ .../agentplugins/conformance/scan.go | 22 ++- 3 files changed, 256 insertions(+), 28 deletions(-) create mode 100644 install/integrationctl/agentplugins/conformance/root_compat_test.go diff --git a/cli/plugin-kit-ai/internal/authoring/commands/slice_test.go b/cli/plugin-kit-ai/internal/authoring/commands/slice_test.go index e20b6f5a..f1b419cd 100644 --- a/cli/plugin-kit-ai/internal/authoring/commands/slice_test.go +++ b/cli/plugin-kit-ai/internal/authoring/commands/slice_test.go @@ -14,8 +14,8 @@ import ( "reflect" "runtime" "strings" - "sync" "testing" + "time" "github.com/777genius/plugin-kit-ai/cli/internal/agentpluginscli" "github.com/777genius/plugin-kit-ai/cli/internal/authoring/commands" @@ -71,6 +71,16 @@ func decodeReport(t *testing.T, b []byte) report.Report { } func execute(t *testing.T, a commands.App, args []string, mount bool) (report.Report, int, []byte) { t.Helper() + return decodeExecution(t, executeRaw(a, args, mount)) +} + +// Workers capture bytes and errors; only the parent may decode or fail a test. +type rawExecution struct { + out, errout []byte + err error +} + +func executeRaw(a commands.App, args []string, mount bool) rawExecution { var out, errout bytes.Buffer builder := commands.RootBuilder(authoringcli.NewPluginKitRoot) if mount { @@ -78,15 +88,75 @@ func execute(t *testing.T, a commands.App, args []string, mount bool) (report.Re args = append([]string{"author"}, args...) } e := a.Execute(context.Background(), args, authoringcli.Streams{Out: &out, Err: &errout}, builder) - if errout.Len() != 0 { - t.Fatalf("duplicate stderr %q", errout.String()) + return rawExecution{out.Bytes(), errout.Bytes(), e} +} + +func decodeExecution(t *testing.T, result rawExecution) (report.Report, int, []byte) { + t.Helper() + if len(result.errout) != 0 { + t.Fatalf("duplicate stderr %q", result.errout) } code := 0 - if e != nil { - code = exitx.Code(e) + if result.err != nil { + code = exitx.Code(result.err) + } + return decodeReport(t, result.out), code, result.out +} + +// Drain every worker before assertions so a parent Fatal cannot strand work. +func concurrentResults(jobs ...func() rawExecution) []rawExecution { + start := make(chan struct{}) + results := make(chan rawExecution, len(jobs)) + for _, job := range jobs { + go func() { + <-start + results <- job() + }() } - return decodeReport(t, out.Bytes()), code, out.Bytes() + close(start) + collected := make([]rawExecution, 0, len(jobs)) + for range jobs { + collected = append(collected, <-results) + } + return collected } + +func TestConcurrentResultsFaults(t *testing.T) { + const faultEnv = "UAP_POSTMERGE_CONCURRENT_FAULT" + if fault := os.Getenv(faultEnv); fault != "" { + results := concurrentResults(func() rawExecution { + if fault == "stderr" { + return rawExecution{out: []byte(`{}`), errout: []byte("injected stderr"), err: fmt.Errorf("injected command error")} + } + return rawExecution{out: []byte(`{`), err: fmt.Errorf("injected command error")} + }, func() rawExecution { return rawExecution{out: []byte(`{}`)} }) + for _, result := range results { + decodeExecution(t, result) + } + return + } + for _, tc := range []struct{ fault, diagnostic string }{ + {"malformed", "invalid report:"}, {"stderr", "duplicate stderr"}, + } { + t.Run(tc.fault, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestConcurrentResultsFaults$", "-test.timeout=8s") + cmd.Env = append(os.Environ(), faultEnv+"="+tc.fault) + out, err := cmd.CombinedOutput() + if ctx.Err() != nil || err == nil || !bytes.Contains(out, []byte(tc.diagnostic)) || bytes.Contains(out, []byte("test timed out")) { + t.Fatalf("fault did not fail promptly: %v (%v)\n%s", err, ctx.Err(), out) + } + }) + } + results := concurrentResults(func() rawExecution { + return rawExecution{out: []byte(`{}`), err: fmt.Errorf("injected command error")} + }) + if _, code, _ := decodeExecution(t, results[0]); code == 0 { + t.Fatal("worker command error lost") + } +} + func write(t *testing.T, root, path, body string) { t.Helper() p := filepath.Join(root, filepath.FromSlash(path)) @@ -233,18 +303,18 @@ func TestReportsAndFreshFactory(t *testing.T) { } root := t.TempDir() write(t, root, "plugin.json", plugin("")) - var wg sync.WaitGroup + var jobs []func() rawExecution for i := 0; i < 8; i++ { - wg.Add(1) - go func(i int) { - defer wg.Done() - r, c, _ := execute(t, a, []string{"validate", root, "--format=json"}, i%2 == 0) - if c != 0 || r.Conformance.Status != report.Pass { - t.Error("concurrent factory failed") - } - }(i) + jobs = append(jobs, func() rawExecution { + return executeRaw(a, []string{"validate", root, "--format=json"}, i%2 == 0) + }) + } + for _, result := range concurrentResults(jobs...) { + r, c, _ := decodeExecution(t, result) + if c != 0 || r.Conformance.Status != report.Pass { + t.Error("concurrent factory failed") + } } - wg.Wait() } func TestArgumentFailuresBeforeEffects(t *testing.T) { @@ -334,19 +404,13 @@ func TestConcurrentInitAndCanceledInvocation(t *testing.T) { a := commands.App{Projects: project.Service{Scratch: scratch}, Revision: baseline} dest := filepath.Join(parent, "demo") args := []string{"init", dest, "--template=skill", "--name=demo", "--description=A fixture.", "--format=json"} - start := make(chan struct{}) - results := make(chan report.Report, 2) + var jobs []func() rawExecution for i := 0; i < 2; i++ { - go func(mount bool) { - <-start - r, _, _ := execute(t, a, args, mount) - results <- r - }(i == 1) + jobs = append(jobs, func() rawExecution { return executeRaw(a, args, i == 1) }) } - close(start) wins := 0 - for i := 0; i < 2; i++ { - r := <-results + for _, result := range concurrentResults(jobs...) { + r, _, _ := decodeExecution(t, result) if r.Committed { wins++ } else if r.Error == nil || r.Error.Code != "destination_exists" { diff --git a/install/integrationctl/agentplugins/conformance/root_compat_test.go b/install/integrationctl/agentplugins/conformance/root_compat_test.go new file mode 100644 index 00000000..abb76646 --- /dev/null +++ b/install/integrationctl/agentplugins/conformance/root_compat_test.go @@ -0,0 +1,146 @@ +package conformance + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + + "github.com/777genius/plugin-kit-ai/install/integrationctl/agentplugins/domain" +) + +func opaqueRootFixture() []byte { + var b strings.Builder + b.WriteString(`{"$schema":"` + domain.PluginSchemaV1 + `","name":"good","future":{`) + for i := 0; i < 60000; i++ { + if i > 0 { + b.WriteByte(',') + } + fmt.Fprintf(&b, `"k%05d":0`, i) + } + b.WriteString(`}}`) + return []byte(b.String()) +} + +func TestRootCompatibilityAllocations(t *testing.T) { + body := opaqueRootFixture() + var scanErr error + allocs := testing.AllocsPerRun(3, func() { + _, scanErr = scanJSON(context.Background(), body, Limits{}, duplicateRoot) + }) + if scanErr != nil { + t.Fatal(scanErr) + } + // Allow generous runtime variation, but no allocation per opaque member. + if allocs > 2000 { + t.Fatalf("root scan allocated %.0f times for %d bytes; ceiling 2000", allocs, len(body)) + } + t.Logf("root scan: bytes=%d allocations=%.0f", len(body), allocs) +} + +func BenchmarkRootCompatibility(b *testing.B) { + body := opaqueRootFixture() + b.ReportAllocs() + b.SetBytes(int64(len(body))) + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := scanJSON(context.Background(), body, Limits{}, duplicateRoot); err != nil { + b.Fatal(err) + } + } +} + +func TestRootCompatibilityStructure(t *testing.T) { + for _, tc := range []struct{ name, body, diagnostic string }{ + {"nested-object", `{"future":{"x":1,"x":2}}`, ""}, + {"nested-array", `{"future":[{"x":1,"x":2}]}`, ""}, + {"root-escaped-duplicate", `{"x":1,"\u0078":2}`, `JSON object contains duplicate field "x"`}, + {"root-array", `[]`, "document must be a JSON object"}, + {"root-null", `null`, "document must be a JSON object"}, + {"trailing", `{} {}`, "document contains multiple JSON values"}, + {"nested-syntax", `{"future":{"x":}}`, "invalid character"}, + {"unterminated", `{"future":[1`, "unexpected EOF"}, + {"trailing-syntax", `{} !`, "invalid character"}, + } { + t.Run(tc.name, func(t *testing.T) { + for _, check := range []func([]byte) error{ + rejectDuplicateTopLevelObjectKeys, + func(b []byte) error { _, _, e := DecodeJSONObject(b); return e }, + func(b []byte) error { var raw map[string]json.RawMessage; return DecodeRawJSONObject(b, &raw) }, + } { + err := check([]byte(tc.body)) + if tc.diagnostic == "" { + if err != nil { + t.Fatal(err) + } + } else if err == nil || !strings.Contains(err.Error(), tc.diagnostic) { + t.Fatalf("got %v, want %s", err, tc.diagnostic) + } + } + if tc.diagnostic == "" && RejectDuplicateJSONKeys([]byte(tc.body)) == nil { + t.Fatal("all-key checking lost nested duplicate") + } + }) + } + for _, depth := range []int{10000, 10001} { + body := []byte(`{"future":` + strings.Repeat("[", depth-1) + "0" + strings.Repeat("]", depth-1) + "}") + err := rejectDuplicateTopLevelObjectKeys(body) + if (err == nil) != (depth == 10000) { + t.Fatalf("depth %d: %v", depth, err) + } + } + // Explicit bounds must still traverse descendants, even with root policy. + for _, limits := range []Limits{{Tokens: 3}, {Members: 1}, {Depth: 1}} { + if _, err := scanJSON(context.Background(), []byte(`{"future":{"x":1}}`), limits, duplicateRoot); err == nil { + t.Fatalf("ignored limits: %+v", limits) + } + } +} + +func TestRootCompatibilityInstallerAndAuthorPolicies(t *testing.T) { + d := decoder(t) + installer := InstallerDecoder{Registry: d.Registry} + for _, field := range []string{"future", "extensions"} { + body := []byte(`{"$schema":"` + domain.PluginSchemaV1 + `","name":"good","` + field + `":{"example.fixture":{"x":1,"x":2}}}`) + _, diagnostics, _, err := installer.Plugin(body) + if field == "extensions" { + if err == nil || !strings.Contains(err.Error(), `duplicate field "x"`) { + t.Fatalf("known field duplicate: %v", err) + } + } else if err != nil || len(diagnostics) != 1 || diagnostics[0].Code != "plugin_unknown_field" || diagnostics[0].Item != "future" { + t.Fatalf("opaque duplicate disposition: %v %+v", err, diagnostics) + } + } + for _, tc := range []struct{ code, value string }{ + {"document_member_limit", func() string { + var b strings.Builder + b.WriteByte('{') + for i := 0; i <= DefaultLimits().Members; i++ { + if i > 0 { + b.WriteByte(',') + } + fmt.Fprintf(&b, `"k%d":0`, i) + } + b.WriteByte('}') + return b.String() + }()}, + {"document_token_limit", "[" + strings.Repeat("0,", DefaultLimits().Tokens) + "0]"}, + } { + t.Run(tc.code, func(t *testing.T) { + body := []byte(`{"$schema":"` + domain.PluginSchemaV1 + `","name":"good","future":` + tc.value + `}`) + if _, _, err := DecodeJSONObject(body); err != nil { + t.Fatal(err) + } + if _, diagnostics, _, err := installer.Plugin(body); err != nil || len(diagnostics) != 1 || diagnostics[0].Code != "plugin_unknown_field" { + t.Fatalf("compatibility imposed author bounds: %v %+v", err, diagnostics) + } + input := minimal() + input.Plugin = NewDocument("plugin.json", Present, body) + f, err := d.Decode(context.Background(), input) + if err != nil || f.Conformance != NotEvaluated || f.Coverage.Complete || !hasFinding(f, tc.code, HostSafety, domain.BoundaryPlugin) { + t.Fatalf("author preflight changed: %+v %v", f, err) + } + }) + } +} diff --git a/install/integrationctl/agentplugins/conformance/scan.go b/install/integrationctl/agentplugins/conformance/scan.go index 2af905a9..c6f40fae 100644 --- a/install/integrationctl/agentplugins/conformance/scan.go +++ b/install/integrationctl/agentplugins/conformance/scan.go @@ -25,16 +25,18 @@ type parseFailure struct { func (e *parseFailure) Error() string { return e.Code } func (e *parseFailure) Unwrap() error { return e.cause } -// scanJSON is the single structural token walk. Compatibility callers select +// scanJSON checks JSON structure. Compatibility callers select // their old duplicate boundary and no new limits; author callers record bounded // duplicate locations before any map materialization. func scanJSON(ctx context.Context, body []byte, limits Limits, mode duplicateMode) ([]duplicate, error) { + // Only unbounded root compatibility checks may skip opaque descendants. + skipRootValues := mode == duplicateRoot && limits == (Limits{}) // encoding/json map decoding already rejects nesting beyond 10000. Keep // that existing ceiling in the compatibility walk before recursive descent. if limits.Depth == 0 { limits.Depth = 10000 } - s := jsonScanner{ctx: ctx, decoder: json.NewDecoder(bytes.NewReader(body)), limits: limits, mode: mode} + s := jsonScanner{ctx: ctx, decoder: json.NewDecoder(bytes.NewReader(body)), limits: limits, mode: mode, skipRootValues: skipRootValues} s.decoder.UseNumber() token, err := s.token() if err != nil { @@ -52,6 +54,11 @@ func scanJSON(ctx context.Context, body []byte, limits Limits, mode duplicateMod } return nil, err } + // RawMessage decoding counts depth from the value, excluding the root. + // Validate the whole document too to retain encoding/json's 10000 ceiling. + if skipRootValues && !json.Valid(body) { + return nil, &parseFailure{Code: "document_depth_limit"} + } return s.duplicates, nil } @@ -60,6 +67,7 @@ type jsonScanner struct { decoder *json.Decoder limits Limits mode duplicateMode + skipRootValues bool tokens, members int duplicates []duplicate } @@ -111,6 +119,16 @@ func (s *jsonScanner) value(token json.Token, keys []string, depth int) error { } } seen[key] = true + if s.skipRootValues { + if err := s.ctx.Err(); err != nil { + return err + } + var value json.RawMessage + if err := s.decoder.Decode(&value); err != nil { + return err + } + continue + } t, err = s.token() if err != nil { return err