777genius · 777genius · Sep 6, 2026 · Sep 6, 2026
@@ -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,22 +71,92 @@ 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 {
 		builder = mounted
 		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" {

@@ -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)
+			}
+		})
+	}
+}
@@ -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