Reject repository prefix collisions without blocking registry access

bench/load/call.go

@@ -54,8 +54,10 @@ func checkToolResult(data json.RawMessage) error {
 		return errors.New("tool returned isError without error content")
 	}
 	var body struct {
-		Code    string `json:"error_code"`
-		Message string `json:"message"`
+		Code      string         `json:"error_code"`
+		Message   string         `json:"message"`
+		Retriable bool           `json:"retriable"`
+		Data      map[string]any `json:"data"`
 	}
 	if err := json.Unmarshal(payload, &body); err != nil {
 		return err
@@ -64,7 +66,7 @@ func checkToolResult(data json.RawMessage) error {
 	if body.Code == "invalid_argument" {
 		code = -32602
 	}
-	return &rpcError{Code: code, Message: body.Code + ": " + body.Message}
+	return &rpcError{Code: code, Message: body.Code + ": " + body.Message, Data: body.Data, Retriable: body.Code == "repo_loading" && body.Retriable}
 }
 
 type toolCaller interface {
@@ -74,6 +76,14 @@ type toolCaller interface {
 // Each logical call has its own product budget. The phase context may end
 // sooner; that observation is right-censored, never a product deadline.
 func callTool(ctx context.Context, client toolCaller, r request, args map[string]any, role string, agent, step, slot int, timeout time.Duration, sleep func(context.Context, time.Duration) error) (json.RawMessage, []Attempt, error) {
+	return callToolPolicy(ctx, client, r, args, role, agent, step, slot, timeout, sleep, false)
+}
+
+func callEditProbeTool(ctx context.Context, client toolCaller, r request, args map[string]any, role string, agent, step, slot int, timeout time.Duration, sleep func(context.Context, time.Duration) error) (json.RawMessage, []Attempt, error) {
+	return callToolPolicy(ctx, client, r, args, role, agent, step, slot, timeout, sleep, true)
+}
+
+func callToolPolicy(ctx context.Context, client toolCaller, r request, args map[string]any, role string, agent, step, slot int, timeout time.Duration, sleep func(context.Context, time.Duration) error, retryTyped bool) (json.RawMessage, []Attempt, error) {
 	logicalStart := time.Now()
 	productEnd := logicalStart.Add(timeout)
 	phaseEnd, phaseBounded := ctx.Deadline()
@@ -94,28 +104,47 @@ func callTool(ctx context.Context, client toolCaller, r request, args map[string
 	for retry := 0; ; retry++ {
 		start := time.Now()
 		data, err := client.call(callCtx, "tools/call", map[string]any{"name": r.Tool, "arguments": args})
+		returned := time.Now()
+		if retryTyped {
+			if clock, ok := client.(interface{ responseReturnedAt() time.Time }); ok {
+				if observed := clock.responseReturnedAt(); !observed.IsZero() && !observed.Before(start) {
+					returned = observed
+				}
+			}
+		}
 		if err == nil && callCtx.Err() != nil {
 			err = callCtx.Err()
 		}
 		if err == nil {
 			err = checkToolResult(data)
 		}
-		a := Attempt{Tool: r.name(), Role: role, Request: hash, Agent: agent, Step: step, Slot: slot, Retry: retry, Started: start, DurationMS: float64(time.Since(start)) / float64(time.Millisecond), LogicalMS: float64(time.Since(logicalStart)) / float64(time.Millisecond), Outcome: "success"}
+		a := Attempt{Tool: r.name(), Role: role, Request: hash, Agent: agent, Step: step, Slot: slot, Retry: retry, Started: start, DurationMS: float64(returned.Sub(start)) / float64(time.Millisecond), LogicalMS: float64(returned.Sub(logicalStart)) / float64(time.Millisecond), Outcome: "success"}
 		var remote *rpcError
 		if err != nil {
 			a.Outcome, a.Error = "error", err.Error()
 			if errors.As(err, &remote) {
 				a.Code = remote.Code
 			}
-			if remote != nil && remote.Code == -32002 {
+			if remote != nil && remote.Code == -32002 && (!retryTyped || remote.retryEditProbe()) {
 				a.Outcome = "busy"
+			} else if retryTyped && remote != nil && remote.retryEditProbe() {
+				a.Outcome = "retryable"
+			}
+			if _, expected := err.(*editProbeExpectedAbsence); expected {
+				a.Outcome = "expected_refusal"
+				a.Code = -32000
+				a.ExpectedOutcome = "symbol_absent"
 			}
 			if callCtx.Err() != nil {
 				a.Outcome = terminalOutcome()
+				err = errors.Join(err, callCtx.Err())
+			}
+			if retryTyped && errors.Is(err, errEditProbeEvidenceIncomplete) {
+				a.Outcome = "harness_incomplete"
 			}
 		}
 		attempts = append(attempts, a)
-		if a.Outcome != "busy" {
+		if a.Outcome != "busy" && a.Outcome != "retryable" {
 			return data, attempts, err
 		}
 		if err := sleep(callCtx, busyBackoff(retry)); err != nil {

bench/load/client.go

@@ -17,12 +17,27 @@ import (
 const maxWireBytes = 64 << 20
 
 type rpcError struct {
-	Code    int    `json:"code"`
-	Message string `json:"message"`
+	Code      int            `json:"code"`
+	Message   string         `json:"message"`
+	Data      map[string]any `json:"data,omitempty"`
+	Retriable bool           `json:"-"` // typed tool-result refusal, not a JSON-RPC top-level field
 }
 
 func (e *rpcError) Error() string { return fmt.Sprintf("RPC %d: %s", e.Code, e.Message) }
 
+func (e *rpcError) retryEditProbe() bool {
+	if unknown, _ := e.Data["delivery_unknown"].(bool); unknown {
+		return false
+	}
+	// These are actual front-door shapes: dispatcher -32002 in
+	// mcp_request_lifetime.go, and repo_loading data in
+	// cmd/gortex/daemon_mcp_repo_loading.go. Unknowns stay terminal.
+	if e.Code == -32002 {
+		return true
+	}
+	return e.Retriable || e.Code == -32000 && e.Data["error_code"] == "repo_loading" && e.Data["retryable"] == true
+}
+
 type wireReply struct {
 	ID     json.Number     `json:"id"`
 	Result json.RawMessage `json:"result"`
@@ -51,6 +66,14 @@ type socketClient struct {
 	version       string
 }
 
+// Per-call values returned to its caller, never a shared last-ID cache. Times
+// retain monotonic readings until the scenario renders relative nanoseconds.
+type wireCallIdentity struct {
+	ID                                                             int64
+	onWriteStart                                                   func(time.Time)
+	Started, WriteStarted, WriteFinished, ReplyDelivered, Finished time.Time
+}
+
 func readWireLine(reader *bufio.Reader) ([]byte, error) {
 	var data []byte
 	for {
@@ -66,6 +89,12 @@ func readWireLine(reader *bufio.Reader) ([]byte, error) {
 }
 
 func dialSocket(ctx context.Context, socket, cwd, mode string) (net.Conn, *bufio.Reader, string, error) {
+	return dialSocketSurface(ctx, socket, cwd, mode, "facade-v1", "hide")
+}
+
+// dialSocketSurface negotiates the session's tool surface in the handshake,
+// as the CLI does per connection; the load workload always uses the facade.
+func dialSocketSurface(ctx context.Context, socket, cwd, mode, tools, toolsMode string) (net.Conn, *bufio.Reader, string, error) {
 	conn, err := (&net.Dialer{}).DialContext(ctx, "unix", socket)
 	if err != nil {
 		return nil, nil, "", err
@@ -80,7 +109,7 @@ func dialSocket(ctx context.Context, socket, cwd, mode string) (net.Conn, *bufio
 	}
 	stop := context.AfterFunc(ctx, func() { _ = conn.Close() })
 	defer stop()
-	handshake := map[string]any{"version": 1, "mode": mode, "cwd": cwd, "client": "gortex-load", "tools": "facade-v1", "tools_mode": "hide"}
+	handshake := map[string]any{"version": 1, "mode": mode, "cwd": cwd, "client": "gortex-load", "tools": tools, "tools_mode": toolsMode}
 	if err := json.NewEncoder(conn).Encode(handshake); err != nil {
 		_ = conn.Close()
 		return nil, nil, "", err
@@ -190,6 +219,10 @@ func (c *socketClient) Close() error {
 }
 
 func (c *socketClient) send(ctx context.Context, value any, advisory bool) error {
+	return c.sendObserved(ctx, value, advisory, nil)
+}
+
+func (c *socketClient) sendObserved(ctx context.Context, value any, advisory bool, observation *wireCallIdentity) error {
 	data, err := json.Marshal(value)
 	if err != nil {
 		return err
@@ -227,7 +260,16 @@ func (c *socketClient) send(ctx context.Context, value any, advisory bool) error
 		_ = c.conn.Close()
 		close(interrupted)
 	})
+	if observation != nil {
+		observation.WriteStarted = time.Now()
+		if observation.onWriteStart != nil {
+			observation.onWriteStart(observation.WriteStarted)
+		}
+	}
 	_, err = io.Copy(c.conn, bytes.NewReader(append(data, '\n')))
+	if observation != nil {
+		observation.WriteFinished = time.Now()
+	}
 	if !stop() {
 		<-interrupted
 	}
@@ -242,6 +284,26 @@ func (c *socketClient) send(ctx context.Context, value any, advisory bool) error
 }
 
 func (c *socketClient) call(ctx context.Context, method string, params any) (json.RawMessage, error) {
+	return c.callObserved(ctx, method, params, nil)
+}
+
+func (c *socketClient) callWithIdentity(ctx context.Context, method string, params any) (json.RawMessage, wireCallIdentity, error) {
+	return c.callWithWriteObserver(ctx, method, params, nil)
+}
+
+func (c *socketClient) callWithWriteObserver(ctx context.Context, method string, params any, onWriteStart func(time.Time)) (json.RawMessage, wireCallIdentity, error) {
+	var observation wireCallIdentity
+	observation.onWriteStart = onWriteStart
+	data, err := c.callObserved(ctx, method, params, &observation)
+	observation.onWriteStart = nil
+	return data, observation, err
+}
+
+func (c *socketClient) callObserved(ctx context.Context, method string, params any, observation *wireCallIdentity) (json.RawMessage, error) {
+	if observation != nil {
+		observation.Started = time.Now()
+		defer func() { observation.Finished = time.Now() }()
+	}
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
@@ -253,6 +315,9 @@ func (c *socketClient) call(ctx context.Context, method string, params any) (jso
 	}
 	c.next++
 	id := c.next
+	if observation != nil {
+		observation.ID = id
+	}
 	ch := make(chan wireReply, 1)
 	c.pending[id] = ch
 	c.mu.Unlock()
@@ -261,7 +326,7 @@ func (c *socketClient) call(ctx context.Context, method string, params any) (jso
 		delete(c.pending, id)
 		c.mu.Unlock()
 	}()
-	if err := c.send(ctx, map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}, false); err != nil {
+	if err := c.sendObserved(ctx, map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}, false, observation); err != nil {
 		return nil, err
 	}
 	select {
@@ -273,6 +338,9 @@ func (c *socketClient) call(ctx context.Context, method string, params any) (jso
 		cancel()
 		return nil, ctx.Err()
 	case reply := <-ch:
+		if observation != nil {
+			observation.ReplyDelivered = time.Now()
+		}
 		if ctx.Err() != nil {
 			return reply.Result, ctx.Err()
 		}

bench/load/client_test.go

@@ -427,3 +427,63 @@ func TestInvalidationSubscriptionRecordsAvailabilityAndActualArguments(t *testin
 		})
 	}
 }
+
+func TestRecordedCallsKeepExactIdentityAcrossMultiplexedReplies(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	socket := startStub(t, func(frame stubFrame) (any, *rpcError) {
+		return map[string]any{"wire_id": frame.ID, "ordinal": frame.Params["ordinal"]}, nil
+	})
+	client, err := openMCP(ctx, socket, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	type result struct {
+		ordinal  int
+		data     json.RawMessage
+		identity wireCallIdentity
+		err      error
+	}
+	results := make(chan result, 32)
+	var calls sync.WaitGroup
+	for ordinal := range 32 {
+		calls.Go(func() {
+			data, id, err := client.callWithIdentity(ctx, "identity", map[string]any{"ordinal": ordinal})
+			results <- result{ordinal, data, id, err}
+		})
+	}
+	calls.Wait()
+	close(results)
+	seen := make(map[int64]bool)
+	for result := range results {
+		if result.err != nil {
+			t.Fatal(result.err)
+		}
+		var decoded struct {
+			WireID  int64 `json:"wire_id"`
+			Ordinal int   `json:"ordinal"`
+		}
+		if err := json.Unmarshal(result.data, &decoded); err != nil {
+			t.Fatal(err)
+		}
+		id := result.identity
+		if id.ID != decoded.WireID || decoded.Ordinal != result.ordinal || seen[id.ID] {
+			t.Fatalf("misattributed multiplexed reply%+v %+v", result, decoded)
+		}
+		seen[id.ID] = true
+		if id.Started.IsZero() || id.WriteStarted.Before(id.Started) || id.WriteFinished.Before(id.WriteStarted) || id.ReplyDelivered.Before(id.WriteFinished) || id.Finished.Before(id.ReplyDelivered) {
+			t.Fatalf("incomplete/reversed call readings%+v", id)
+		}
+	}
+}
+
+func TestRecordedCanceledCallDoesNotFabricateWireOrReply(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	client := &socketClient{}
+	data, id, err := client.callWithIdentity(ctx, "not-sent", nil)
+	if !errors.Is(err, context.Canceled) || data != nil || id.ID != 0 || !id.WriteStarted.IsZero() || !id.ReplyDelivered.IsZero() || id.Started.IsZero() || id.Finished.Before(id.Started) {
+		t.Fatalf("fabricated canceled call data=%s identity=%+v err=%v", data, id, err)
+	}
+}