package openai

import (
	"context"
	"encoding/json"
	"errors"
	"net"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"

	edgenode "iop/apps/edge/internal/node"
	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
	"iop/packages/go/singlerequesttemplate"
	iop "iop/proto/gen/iop"
)

type reviewController struct {
	mu           sync.Mutex
	binding      *edgeservice.SingleRequestBinding
	plan         []byte
	review       []byte
	state        edgeservice.SingleRequestState
	envelopes    []edgeservice.SingleRequestEnvelope
	writes       []edgeservice.SingleRequestArtifactKind
	artifact     []byte
	bridge       *singleRequestWorkToolBridge
	autoContinue bool
	toolResult   func(*edgeservice.InternalWorkspaceToolCall) edgeservice.InternalWorkspaceToolResult
	writeErr     error
	envelopeErr  error
}

func (c *reviewController) RequestID() string                          { return "request-review" }
func (c *reviewController) Binding() *edgeservice.SingleRequestBinding { return c.binding.Clone() }
func (c *reviewController) Context() context.Context                   { return context.Background() }
func (c *reviewController) State() edgeservice.SingleRequestState {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.state
}
func (c *reviewController) ReadInternalArtifact(_ context.Context, kind edgeservice.SingleRequestArtifactKind) ([]byte, error) {
	switch kind {
	case edgeservice.SingleRequestArtifactPlan:
		return append([]byte(nil), c.plan...), nil
	case edgeservice.SingleRequestArtifactReview:
		return append([]byte(nil), c.review...), nil
	default:
		return nil, errors.New("unexpected artifact")
	}
}
func (c *reviewController) WriteInternalArtifact(_ context.Context, kind edgeservice.SingleRequestArtifactKind, content []byte) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.writeErr != nil {
		return c.writeErr
	}
	c.writes = append(c.writes, kind)
	c.artifact = append([]byte(nil), content...)
	return nil
}
func (c *reviewController) SubmitEnvelope(env edgeservice.SingleRequestEnvelope) error {
	c.mu.Lock()
	if c.envelopeErr != nil {
		c.mu.Unlock()
		return c.envelopeErr
	}
	c.envelopes = append(c.envelopes, env)
	c.state = env.Stage
	auto, bridge, toolResult := c.autoContinue, c.bridge, c.toolResult
	c.mu.Unlock()
	if auto && env.Stage == edgeservice.SingleRequestStateInternalTool {
		go func(call *edgeservice.InternalWorkspaceToolCall) {
			result := edgeservice.InternalWorkspaceToolResult{RequestID: call.RequestID, StageID: call.StageID, ToolCallID: call.ToolCallID, Status: "success", Stdout: []byte("inspection complete")}
			if toolResult != nil {
				result = toolResult(call)
			}
			_ = bridge.ContinueInternalTool(context.Background(), result)
		}(env.ToolCall.Clone())
	}
	return nil
}

func reviewRequest(t *testing.T) singleRequestReviewStageRequest {
	t.Helper()
	binding := workBinding(t)
	return singleRequestReviewStageRequest{RequestID: "request-review", Task: "update file", StageBinding: binding.Review, Limits: binding.Limits, NodeRef: "node", SessionID: "review-session", UsageAttribution: "principal", Sequence: 3}
}

func reviewPassBody(output, summary string) []byte {
	decision, _ := json.Marshal(map[string]string{
		"decision":     "pass",
		"output":       output,
		"checks":       "- Checked requirements",
		"verification": "- Verified tests pass",
		"summary":      summary,
	})
	b, _ := json.Marshal(map[string]any{"id": "id", "object": "chat.completion", "created": 1, "model": "gemini-3.6-flash", "choices": []any{map[string]any{"index": 0, "finish_reason": "stop", "message": map[string]any{"role": "assistant", "content": string(decision), "reasoning_content": "provider-private-review-reasoning", "extra_content": map[string]any{"google": map[string]any{"thought_signature": "provider-private-final-signature"}}}}}})
	return b
}

func reviewToolBody(id, name, args string) []byte {
	b, _ := json.Marshal(map[string]any{
		"id": "id", "object": "chat.completion", "created": 1, "model": "gemini-3.6-flash",
		"choices": []any{map[string]any{
			"index": 0, "finish_reason": "tool_calls",
			"message": map[string]any{
				"role": "assistant", "content": nil, "reasoning_content": "provider-private-review-reasoning",
				"tool_calls": []any{map[string]any{
					"id": id, "type": "function", "function": map[string]any{"name": name, "arguments": args},
					"extra_content": map[string]any{"google": map[string]any{"thought_signature": "provider-private-tool-signature"}},
				}},
			},
		}},
	})
	return b
}

func newReviewController(t *testing.T, bridge *singleRequestWorkToolBridge) *reviewController {
	t.Helper()
	return &reviewController{binding: workBinding(t), plan: []byte("# Plan\n\n## Goal\nUpdate result.\n\n## Steps\n- [P1] Write result.txt.\n- [P2] Verify result.\n\n## Verification\n- Run verify.\n"), review: []byte("# Review\n\n## Worker Item Status\n- P1: completed\n- P2: completed\n\n## Worker Changes\nUpdated result.txt.\n\n## Worker Verification\nverify passed\n\n## Deviations\nNone\n"), state: edgeservice.SingleRequestStateWorking, bridge: bridge, autoContinue: true}
}

func scriptedReviewProvider(t *testing.T, ctrl *reviewController, bodies [][]byte, captured *[][]byte) *singleRequestProviderStage {
	t.Helper()
	var mu sync.Mutex
	index := 0
	return newSingleRequestProviderStage(&mockService{submit: func(_ context.Context, request edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
		body, err := request.Tunnel.BuildBody("gemini-3.6-flash")
		if err != nil {
			return nil, err
		}
		mu.Lock()
		*captured = append(*captured, body)
		current := index
		index++
		mu.Unlock()
		if current >= len(bodies) {
			return nil, errors.New("unexpected provider call")
		}
		return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(bodies[current])}, DispatchInfo: matchingDispatch()}, nil
	}})
}

type reviewStageExecutionOutcome struct {
	result *singleRequestReviewResult
	err    error
}

type serviceReviewStageExecutor struct {
	stage            *singleRequestReviewStage
	plan             []byte
	outcomes         chan reviewStageExecutionOutcome
	continueCount    atomic.Int32
	reviewWriteCount atomic.Int32
	finalizingCount  atomic.Int32
}

func (e *serviceReviewStageExecutor) ExecuteSingleRequest(ctx context.Context, req edgeservice.SingleRequestRequest, ctrl edgeservice.SingleRequestController) error {
	tracked := &reviewSequenceController{SingleRequestController: ctrl, executor: e}
	if err := tracked.SubmitEnvelope(edgeservice.SingleRequestEnvelope{RequestID: req.RequestID, Sequence: 1, Stage: edgeservice.SingleRequestStatePlanning}); err != nil {
		e.outcomes <- reviewStageExecutionOutcome{err: err}
		return err
	}
	if err := tracked.WriteInternalArtifact(ctx, edgeservice.SingleRequestArtifactPlan, e.plan); err != nil {
		e.outcomes <- reviewStageExecutionOutcome{err: err}
		return err
	}
	if err := tracked.WriteInternalArtifact(ctx, edgeservice.SingleRequestArtifactReview, []byte("# Review\n\n## Worker Item Status\n- P1: completed\n- P2: completed\n\n## Worker Changes\nUpdated result.txt.\n\n## Worker Verification\nverify passed\n\n## Deviations\nNone\n")); err != nil {
		e.outcomes <- reviewStageExecutionOutcome{err: err}
		return err
	}
	if err := tracked.SubmitEnvelope(edgeservice.SingleRequestEnvelope{RequestID: req.RequestID, Sequence: tracked.nextSequence(), Stage: edgeservice.SingleRequestStateWorking}); err != nil {
		e.outcomes <- reviewStageExecutionOutcome{err: err}
		return err
	}
	result, err := e.stage.run(ctx, singleRequestReviewStageRequest{
		RequestID:        req.RequestID,
		Task:             req.Prompt,
		StageBinding:     req.Binding.Review,
		Limits:           req.Binding.Limits,
		NodeRef:          req.Binding.Workspace.NodeID,
		SessionID:        "review-stage-test",
		UsageAttribution: "principal-test",
		Sequence:         tracked.nextSequence(),
	}, tracked)
	if err != nil {
		e.outcomes <- reviewStageExecutionOutcome{err: err}
		return err
	}
	e.outcomes <- reviewStageExecutionOutcome{result: result, err: nil}
	return nil
}

func (e *serviceReviewStageExecutor) ContinueInternalTool(ctx context.Context, result edgeservice.InternalWorkspaceToolResult) error {
	err := e.stage.bridge.ContinueInternalTool(ctx, result)
	if err == nil {
		e.continueCount.Add(1)
	}
	return err
}

type reviewSequenceController struct {
	edgeservice.SingleRequestController
	executor *serviceReviewStageExecutor
	mu       sync.Mutex
	last     uint64
}

func (c *reviewSequenceController) WriteInternalArtifact(ctx context.Context, kind edgeservice.SingleRequestArtifactKind, content []byte) error {
	if kind == edgeservice.SingleRequestArtifactReview && c.executor != nil {
		c.executor.reviewWriteCount.Add(1)
	}
	return c.SingleRequestController.WriteInternalArtifact(ctx, kind, content)
}

func (c *reviewSequenceController) SubmitEnvelope(env edgeservice.SingleRequestEnvelope) error {
	if env.Stage == edgeservice.SingleRequestStateFinalizing && c.executor != nil {
		c.executor.finalizingCount.Add(1)
	}
	if err := c.SingleRequestController.SubmitEnvelope(env); err != nil {
		return err
	}
	c.mu.Lock()
	if env.Sequence > c.last {
		c.last = env.Sequence
	}
	c.mu.Unlock()
	return nil
}

func (c *reviewSequenceController) nextSequence() uint64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.last + 1
}

type reviewCoordinatorHarness struct {
	service  *edgeservice.Service
	binding  *edgeservice.SingleRequestBinding
	executor *serviceReviewStageExecutor
	bridge   *singleRequestWorkToolBridge
	node     *workNodeHarness
}

func newReviewCoordinatorHarness(t *testing.T, provider edgeserviceRunner, mutate func(*edgeservice.SingleRequestBinding)) *reviewCoordinatorHarness {
	t.Helper()
	edgeConn, nodeConn := net.Pipe()
	edgeClient := toki.NewTcpClient(edgeConn, 0, 0, toki.ParserMap{
		toki.TypeNameOf(&iop.WorkspaceOpenResponse{}):     parseAnthropicWorkspaceMessage(&iop.WorkspaceOpenResponse{}),
		toki.TypeNameOf(&iop.WorkspaceArtifactResponse{}): parseAnthropicWorkspaceMessage(&iop.WorkspaceArtifactResponse{}),
		toki.TypeNameOf(&iop.WorkspaceToolResponse{}):     parseAnthropicWorkspaceMessage(&iop.WorkspaceToolResponse{}),
		toki.TypeNameOf(&iop.WorkspaceCancelResponse{}):   parseAnthropicWorkspaceMessage(&iop.WorkspaceCancelResponse{}),
		toki.TypeNameOf(&iop.WorkspaceCleanupResponse{}):  parseAnthropicWorkspaceMessage(&iop.WorkspaceCleanupResponse{}),
	})
	nodeClient := toki.NewTcpClient(nodeConn, 0, 0, toki.ParserMap{
		toki.TypeNameOf(&iop.WorkspaceOpenRequest{}):     parseAnthropicWorkspaceMessage(&iop.WorkspaceOpenRequest{}),
		toki.TypeNameOf(&iop.WorkspaceArtifactRequest{}): parseAnthropicWorkspaceMessage(&iop.WorkspaceArtifactRequest{}),
		toki.TypeNameOf(&iop.WorkspaceToolRequest{}):     parseAnthropicWorkspaceMessage(&iop.WorkspaceToolRequest{}),
		toki.TypeNameOf(&iop.WorkspaceCancelRequest{}):   parseAnthropicWorkspaceMessage(&iop.WorkspaceCancelRequest{}),
		toki.TypeNameOf(&iop.WorkspaceCleanupRequest{}):  parseAnthropicWorkspaceMessage(&iop.WorkspaceCleanupRequest{}),
	})
	t.Cleanup(func() {
		_ = edgeClient.Close()
		_ = nodeClient.Close()
	})

	registry := edgenode.NewRegistry()
	registry.Register(&edgenode.NodeEntry{NodeID: "node", Alias: "review-node", Client: edgeClient})
	store := edgenode.NewNodeStore()
	store.Add(&edgenode.NodeRecord{ID: "node", Alias: "review-node", Token: "review-node-token", Workspaces: []config.WorkspaceDefinition{{
		Ref: "workspace", Platform: "darwin", Root: "/Users/operator/project",
		Operations: []config.WorkspaceOperation{config.WorkspaceOpRead, config.WorkspaceOpWrite, config.WorkspaceOpCommand},
		Commands:   []config.WorkspaceCommandDefinition{{ID: "verify", Executable: "/usr/bin/true"}}, EnvironmentAllowlist: []string{"SAFE"},
		MaxReadBytes: 4096, MaxWriteBytes: 4096, MaxOutputBytes: 4096, MaxCommandTimeoutMS: 1000,
	}}})

	binding := workBinding(t)
	binding.Workspace = nil
	binding.Limits.WallClockMS = 5000
	binding.Limits.StageTimeoutMS = 2000
	binding.Limits.MaxToolIterations = 4
	if mutate != nil {
		mutate(binding)
	}
	bridge := newSingleRequestWorkToolBridge()
	executor := &serviceReviewStageExecutor{
		stage:    newSingleRequestReviewStage(newSingleRequestProviderStage(provider), bridge),
		plan:     []byte("# Plan\n\n## Goal\nUpdate result.\n\n## Steps\n- [P1] Write result.txt.\n- [P2] Verify result.\n\n## Verification\n- Run verify.\n"),
		outcomes: make(chan reviewStageExecutionOutcome, 1),
	}
	service := edgeservice.New(registry, nil)
	service.SetNodeStore(store)
	service.SetSingleRequestExecutor(executor)
	nodeHarness := newWorkNodeHarness()
	nodeHarness.install(nodeClient)
	return &reviewCoordinatorHarness{service: service, binding: binding, executor: executor, bridge: bridge, node: nodeHarness}
}

func reviewServiceRequest(binding *edgeservice.SingleRequestBinding) edgeservice.SingleRequestRequest {
	return edgeservice.SingleRequestRequest{
		RequestID: "request-review",
		Binding:   binding,
		Prompt:    "update file",
	}
}

func TestSingleRequestReviewStagePassPersistsBeforeFinalizing(t *testing.T) {
	bridge := newSingleRequestWorkToolBridge()
	ctrl := newReviewController(t, bridge)
	var bodies [][]byte
	stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("Approved output.", "All checks passed.")}, &bodies), bridge)
	result, err := stage.run(context.Background(), reviewRequest(t), ctrl)
	if err != nil {
		t.Fatal(err)
	}
	if string(result.Output) != "Approved output." || result.Summary != "All checks passed." || len(ctrl.writes) != 0 {
		t.Fatalf("result=%+v artifact=%q writes=%v", result, ctrl.artifact, ctrl.writes)
	}
	if len(ctrl.envelopes) != 2 || ctrl.envelopes[0].Stage != edgeservice.SingleRequestStateReviewing || ctrl.envelopes[1].Stage != edgeservice.SingleRequestStateFinalizing || ctrl.envelopes[1].Result == nil || ctrl.envelopes[1].Result.Output != "Approved output." {
		t.Fatalf("envelopes=%+v", ctrl.envelopes)
	}
	if len(bodies) != 1 || !containsAll(string(bodies[0]), singleRequestReviewPrompt, "reasoning_effort", "high", "workspace_read", "Updated result.txt.") {
		t.Fatalf("body=%q", bodies)
	}
}

func TestSingleRequestReviewStageRejectsMismatchedStoredPlan(t *testing.T) {
	t.Run("plan with altered heading fails before provider dispatch", func(t *testing.T) {
		var calls atomic.Int32
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.plan = []byte("# Plan Mismatch\n\n## Goal\nUpdate result.\n\n## Steps\n- [P1] Write result.txt.\n- [P2] Verify result.\n\n## Verification\n- Run verify.\n")
		var bodies [][]byte
		_, err := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("Approved.", "Summary.")}, &bodies), bridge).run(context.Background(), reviewRequest(t), ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if calls.Load() != 0 {
			t.Fatalf("provider dispatches=%d, want 0", calls.Load())
		}
		if len(ctrl.writes) != 0 {
			t.Fatalf("review artifact writes=%d, want 0", len(ctrl.writes))
		}
	})

	t.Run("plan with unknown placeholder fails before provider dispatch", func(t *testing.T) {
		var calls atomic.Int32
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.plan = []byte("# Plan\n\n## Goal\nUpdate result.\n\n## Steps\n- [P1] Write result.txt.\n- [P2] Verify result.\n\n## Verification\n- Run verify.\n{{unknown}}\n")
		var bodies [][]byte
		_, err := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("Approved.", "Summary.")}, &bodies), bridge).run(context.Background(), reviewRequest(t), ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if calls.Load() != 0 {
			t.Fatalf("provider dispatches=%d, want 0", calls.Load())
		}
	})

	t.Run("malformed review handoff with prose in item status fails before provider dispatch", func(t *testing.T) {
		var calls atomic.Int32
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.review = []byte("# Review\n\n## Worker Item Status\n- P1: completed\nThis is a note.\n- P2: completed\n\n## Worker Changes\nUpdated result.txt.\n\n## Worker Verification\nverify passed\n\n## Deviations\nNone\n")
		var bodies [][]byte
		_, err := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("Approved.", "Summary.")}, &bodies), bridge).run(context.Background(), reviewRequest(t), ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if calls.Load() != 0 {
			t.Fatalf("provider dispatches=%d, want 0", calls.Load())
		}
	})
}

func TestSingleRequestReviewStageInspectionAndRepairRemainInLegalStates(t *testing.T) {
	t.Run("inspection", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewToolBody("inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`), reviewPassBody("Approved.", "Inspection passed.")}, &bodies), bridge)
		if _, err := stage.run(context.Background(), reviewRequest(t), ctrl); err != nil {
			t.Fatal(err)
		}
		if bridge.pendingCount() != 0 || len(ctrl.envelopes) != 4 {
			t.Fatalf("pending=%d envelopes=%+v", bridge.pendingCount(), ctrl.envelopes)
		}
		stages := []edgeservice.SingleRequestState{ctrl.envelopes[0].Stage, ctrl.envelopes[1].Stage, ctrl.envelopes[2].Stage, ctrl.envelopes[3].Stage}
		want := []edgeservice.SingleRequestState{edgeservice.SingleRequestStateReviewing, edgeservice.SingleRequestStateInternalTool, edgeservice.SingleRequestStateReviewing, edgeservice.SingleRequestStateFinalizing}
		for i := range want {
			if stages[i] != want[i] {
				t.Fatalf("stages=%v want=%v", stages, want)
			}
		}
		if ctrl.envelopes[1].SavedStage != edgeservice.SingleRequestStateReviewing || len(bodies) != 2 {
			t.Fatalf("tool=%+v bodies=%d", ctrl.envelopes[1], len(bodies))
		}
	})

	t.Run("repair and re-review", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		statesAtDispatch := make([]edgeservice.SingleRequestState, 0, 3)
		stage := newSingleRequestReviewStage(newSingleRequestProviderStage(&mockService{submit: func(_ context.Context, request edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			body, err := request.Tunnel.BuildBody("gemini-3.6-flash")
			if err != nil {
				return nil, err
			}
			bodies = append(bodies, body)
			statesAtDispatch = append(statesAtDispatch, ctrl.State())
			responses := [][]byte{reviewToolBody("repair-1", edgeservice.InternalWorkspaceToolWrite, `{"relative_path":"result.txt","content":"fixed"}`), reviewToolBody("inspect-after-repair", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`), reviewPassBody("Repaired output.", "Repair verified.")}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[len(bodies)-1])}, DispatchInfo: matchingDispatch()}, nil
		}}), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 2
		if _, err := stage.run(context.Background(), req, ctrl); err != nil {
			t.Fatal(err)
		}
		want := []edgeservice.SingleRequestState{edgeservice.SingleRequestStateReviewing, edgeservice.SingleRequestStateRepairing, edgeservice.SingleRequestStateInternalTool, edgeservice.SingleRequestStateRepairing, edgeservice.SingleRequestStateInternalTool, edgeservice.SingleRequestStateRepairing, edgeservice.SingleRequestStateFinalizing}
		if len(ctrl.envelopes) != len(want) {
			t.Fatalf("envelopes=%+v", ctrl.envelopes)
		}
		for i, stage := range want {
			if ctrl.envelopes[i].Stage != stage {
				t.Fatalf("envelopes=%+v", ctrl.envelopes)
			}
		}
		if ctrl.envelopes[2].SavedStage != edgeservice.SingleRequestStateRepairing || ctrl.envelopes[4].SavedStage != edgeservice.SingleRequestStateRepairing || len(statesAtDispatch) != 3 || statesAtDispatch[0] != edgeservice.SingleRequestStateReviewing || statesAtDispatch[1] != edgeservice.SingleRequestStateRepairing || statesAtDispatch[2] != edgeservice.SingleRequestStateRepairing || bridge.pendingCount() != 0 {
			t.Fatalf("dispatch states=%v envelopes=%+v pending=%d", statesAtDispatch, ctrl.envelopes, bridge.pendingCount())
		}
	})
}

func TestSingleRequestReviewStageBoundedInvalidToolCorrection(t *testing.T) {
	t.Run("one malformed provider call is corrected before coordinator admission", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		responses := [][]byte{
			reviewToolBody("invalid-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":""}`),
			reviewToolBody("inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
			reviewPassBody("Approved.", "Corrected inspection passed."),
		}
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, responses, &bodies), bridge)
		if _, err := stage.run(context.Background(), reviewRequest(t), ctrl); err != nil {
			t.Fatal(err)
		}
		if len(bodies) != 3 || !strings.Contains(string(bodies[1]), "invalid_request") {
			t.Fatalf("provider correction bodies=%d", len(bodies))
		}
		want := []edgeservice.SingleRequestState{edgeservice.SingleRequestStateReviewing, edgeservice.SingleRequestStateInternalTool, edgeservice.SingleRequestStateReviewing, edgeservice.SingleRequestStateFinalizing}
		if len(ctrl.envelopes) != len(want) {
			t.Fatalf("envelopes=%+v", ctrl.envelopes)
		}
		for index, state := range want {
			if ctrl.envelopes[index].Stage != state {
				t.Fatalf("envelopes=%+v", ctrl.envelopes)
			}
		}
	})

	t.Run("second malformed provider call fails closed", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		responses := [][]byte{
			reviewToolBody("invalid-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":""}`),
			reviewToolBody("invalid-2", edgeservice.InternalWorkspaceToolRead, `{"relative_path":""}`),
		}
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, responses, &bodies), bridge)
		if _, err := stage.run(context.Background(), reviewRequest(t), ctrl); !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want review stage failure", err)
		}
		if len(bodies) != 2 || len(ctrl.envelopes) != 1 || ctrl.envelopes[0].Stage != edgeservice.SingleRequestStateReviewing || bridge.pendingCount() != 0 {
			t.Fatalf("bodies=%d envelopes=%+v pending=%d", len(bodies), ctrl.envelopes, bridge.pendingCount())
		}
	})
}

func TestSingleRequestReviewStageFailsClosed(t *testing.T) {
	for _, raw := range []string{
		`{"decision":"pass","output":"x","checks":"c","verification":"v","summary":"y","extra":1}`,
		`{"decision":"pass","output":"","checks":"c","verification":"v","summary":"y"}`,
		`{"decision":"pass","output":"x","checks":"","verification":"v","summary":"y"}`,
		`{"decision":"pass","output":"x","checks":"c","verification":"","summary":"y"}`,
		`{"decision":"pass","output":"x","checks":"c","verification":"v","summary":""}`,
		`{"decision":"repair","output":"x","checks":"c","verification":"v","summary":"y"}`,
		`{"decision":"pass","decision":"pass","output":"x","checks":"c","verification":"v","summary":"y"}`,
		`{"decision":"pass","output":"x","summary":"y"}`, // missing checks & verification
		`not-json`,
	} {
		if _, err := decodeSingleRequestReviewDecision(raw, 4096); !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("raw=%q err=%v", raw, err)
		}
	}
	for _, raw := range []string{
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"tool_calls","message":{"role":"assistant","content":null,"tool_calls":[{"id":"a","type":"function","function":{"name":"workspace_read","arguments":"{}"}},{"id":"b","type":"function","function":{"name":"workspace_read","arguments":"{}"}}]}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"{\"decision\":\"pass\",\"output\":\"x\",\"checks\":\"c\",\"verification\":\"v\",\"summary\":\"y\"}","reasoning_content":["private"]}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"{\"decision\":\"pass\",\"output\":\"x\",\"checks\":\"c\",\"verification\":\"v\",\"summary\":\"y\"}","extra_content":null}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"{\"decision\":\"pass\",\"output\":\"x\",\"checks\":\"c\",\"verification\":\"v\",\"summary\":\"y\"}","extra_content":{"google":{"thought_signature":false}}}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"tool_calls","message":{"role":"assistant","content":null,"tool_calls":[{"id":"a","type":"function","function":{"name":"workspace_read","arguments":"{}"},"extra_content":null}]}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"tool_calls","message":{"role":"assistant","content":null,"tool_calls":[{"id":"a","type":"function","function":{"name":"workspace_read","arguments":"{}"},"extra_content":{"google":{"thought_signature":""}}}]}}]}`,
		`{"id":"x","object":"chat.completion","created":1,"model":"m","choices":[{"index":0,"finish_reason":"tool_calls","message":{"role":"assistant","content":null,"tool_calls":[{"id":"a","type":"function","function":{"name":"workspace_read","arguments":"{}"},"extra_content":{"google":{"thought_signature":"sig","unknown":1}}}]}}]}`,
	} {
		if _, err := decodeSingleRequestReviewProviderResponse([]byte(raw), 4096); !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("provider raw=%q err=%v", raw, err)
		}
	}
	t.Run("review does not write an artifact", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.writeErr = errors.New("artifact failure")
		var bodies [][]byte
		_, err := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("Approved.", "Summary.")}, &bodies), bridge).run(context.Background(), reviewRequest(t), ctrl)
		if err != nil || len(ctrl.envelopes) != 2 || len(ctrl.writes) != 0 || bridge.pendingCount() != 0 {
			t.Fatalf("err=%v envelopes=%+v pending=%d", err, ctrl.envelopes, bridge.pendingCount())
		}
	})
	t.Run("bound rejects repeated tool", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 1
		var bodies [][]byte
		_, err := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewToolBody("inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`), reviewToolBody("inspect-2", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`)}, &bodies), bridge).run(context.Background(), req, ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) || len(bodies) != 2 || bridge.pendingCount() != 0 {
			t.Fatalf("err=%v bodies=%d pending=%d", err, len(bodies), bridge.pendingCount())
		}
	})
}

func TestSingleRequestReviewStageCancellationCleansWaiter(t *testing.T) {
	bridge := newSingleRequestWorkToolBridge()
	ctrl := newReviewController(t, bridge)
	ctrl.autoContinue = false
	var bodies [][]byte
	stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewToolBody("inspect-cancel", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`)}, &bodies), bridge)
	ctx, cancel := context.WithCancel(context.Background())
	result := make(chan error, 1)
	go func() { _, err := stage.run(ctx, reviewRequest(t), ctrl); result <- err }()
	deadline := time.After(2 * time.Second)
	for bridge.pendingCount() != 1 {
		select {
		case <-deadline:
			t.Fatal("review tool waiter was not registered")
		case <-time.After(time.Millisecond):
		}
	}
	cancel()
	select {
	case err := <-result:
		if !errors.Is(err, errSingleRequestReviewStage) || bridge.pendingCount() != 0 || len(bodies) != 1 {
			t.Fatalf("err=%v pending=%d bodies=%d", err, bridge.pendingCount(), len(bodies))
		}
	case <-time.After(2 * time.Second):
		t.Fatal("review cancellation did not return")
	}
}

func TestSingleRequestReviewBodyRejectsOptionAliases(t *testing.T) {
	if _, err := buildSingleRequestReviewBody([]chatMessage{{Role: "user", Content: "x"}}, map[string]any{"Reasoning_Effort": "high"}, []any{singleRequestWorkToolSchema(edgeservice.InternalWorkspaceToolRead, map[string]any{"type": "object"})}, "gemini", false); !errors.Is(err, errSingleRequestReviewStage) {
		t.Fatalf("err=%v", err)
	}
	if _, err := buildSingleRequestReviewBody([]chatMessage{{Role: "user", Content: "x"}}, map[string]any{"reasoning_effort": "low"}, []any{singleRequestWorkToolSchema(edgeservice.InternalWorkspaceToolRead, map[string]any{"type": "object"})}, "gemini", false); !errors.Is(err, errSingleRequestReviewStage) {
		t.Fatalf("err=%v", err)
	}
	if _, err := buildSingleRequestReviewBody([]chatMessage{{Role: "user", Content: "x"}}, map[string]any{"reasoning_effort": "high", "Response_Format": map[string]any{"type": "caller"}}, []any{singleRequestWorkToolSchema(edgeservice.InternalWorkspaceToolRead, map[string]any{"type": "object"})}, "gemini", false); !errors.Is(err, errSingleRequestReviewStage) {
		t.Fatalf("response format alias err=%v", err)
	}
	if _, _, err := renderSingleRequestReview(singlerequesttemplate.DefaultReviewTemplate, singleRequestReviewDecision{Decision: "pass", Output: strings.Repeat("x", 10), Checks: "c", Verification: "v", Summary: "summary"}, 9); !errors.Is(err, errSingleRequestReviewStage) {
		t.Fatalf("err=%v", err)
	}
}

func TestSingleRequestReviewResponseFormatIsClosed(t *testing.T) {
	format := singleRequestReviewResponseFormat()
	if format.Type != "json_schema" || format.JSONSchema.Name != "single_request_review" || !format.JSONSchema.Strict {
		t.Fatalf("response format=%+v", format)
	}
	schema := format.JSONSchema.Schema
	if schema.Type != "object" || schema.AdditionalProperties || len(schema.Properties) != 5 || len(schema.Required) != 5 {
		t.Fatalf("review schema=%+v", schema)
	}
	if decision := schema.Properties["decision"]; !reflect.DeepEqual(decision.Enum, []string{"pass"}) {
		t.Fatalf("decision schema=%+v", decision)
	}
	for _, name := range []string{"output", "checks", "verification", "summary"} {
		if schema.Properties[name].MinLength != 1 {
			t.Fatalf("property %q schema=%+v", name, schema.Properties[name])
		}
	}
}

func TestSingleRequestReviewPromptRequiresCanonicalWorkspaceRelativePaths(t *testing.T) {
	for _, requirement := range []string{"canonical", "workspace-relative", "README.md", "./README.md", "absolute path", "parent traversal"} {
		if !strings.Contains(singleRequestReviewPrompt, requirement) {
			t.Fatalf("review prompt does not describe %q path requirement", requirement)
		}
	}
}

// customReviewStageTemplate is an operator-authored effective Review template
// that differs from the built-in default, so a rendered artifact cannot pass by
// accidentally falling back.
const customReviewStageTemplate = `# Review

Operator preamble.

## Result
PASS

## Checks
{{checks}}

## Verification
{{verification}}

## Summary
{{summary}}

Operator footer.
`

// renderSingleRequestReview remains a test-only legacy fixture helper for the
// historical template snapshots below. Production Review no longer renders or
// writes a final REVIEW artifact; the Worker stage owns that handoff.
func renderSingleRequestReview(tmpl string, decision singleRequestReviewDecision, maximum int) ([]byte, *singleRequestReviewResult, error) {
	if decision.Decision != "pass" || maximum < 1 || strings.TrimSpace(decision.Output) == "" {
		return nil, nil, errSingleRequestReviewStage
	}
	artifact := strings.ReplaceAll(tmpl, "{{checks}}", strings.TrimSpace(decision.Checks))
	artifact = strings.ReplaceAll(artifact, "{{verification}}", strings.TrimSpace(decision.Verification))
	artifact = strings.ReplaceAll(artifact, "{{summary}}", strings.TrimSpace(decision.Summary))
	if len(artifact) > maximum {
		return nil, nil, errSingleRequestReviewStage
	}
	return []byte(artifact), &singleRequestReviewResult{Output: []byte(strings.TrimSpace(decision.Output)), Summary: strings.TrimSpace(decision.Summary)}, nil
}

// TestSingleRequestReviewStageCustomTemplateSnapshot proves the Review stage
// renders its internal artifact from the frozen effective template while the
// caller-visible final output stays exactly the model's `decision.output`,
// independent of which template is admitted.
func TestSingleRequestReviewStageCustomTemplateSnapshot(t *testing.T) {
	t.Skip("superseded: REVIEW is a worker handoff and Review no longer renders it")
	decision := singleRequestReviewDecision{
		Decision:     "pass",
		Output:       "  result.txt now contains the requested value.  ",
		Checks:       "- Checked requirements",
		Verification: "- Verified tests pass",
		Summary:      "Task completed.",
	}
	const wantOutput = "result.txt now contains the requested value."

	t.Run("custom template renders the exact internal artifact", func(t *testing.T) {
		artifact, result, err := renderSingleRequestReview(customReviewStageTemplate, decision, 4096)
		if err != nil {
			t.Fatalf("renderSingleRequestReview failed: %v", err)
		}
		want := `# Review

Operator preamble.

## Result
PASS

## Checks
- Checked requirements

## Verification
- Verified tests pass

## Summary
Task completed.

Operator footer.
`
		if string(artifact) != want {
			t.Errorf("artifact:\n%s\nwant:\n%s", string(artifact), want)
		}
		if string(result.Output) != wantOutput {
			t.Errorf("caller output = %q, want %q", string(result.Output), wantOutput)
		}
		if result.Summary != "Task completed." {
			t.Errorf("summary = %q, want %q", result.Summary, "Task completed.")
		}
	})

	t.Run("default template renders the exact internal artifact", func(t *testing.T) {
		artifact, result, err := renderSingleRequestReview(singlerequesttemplate.DefaultReviewTemplate, decision, 4096)
		if err != nil {
			t.Fatalf("renderSingleRequestReview failed: %v", err)
		}
		want := `# Review

## Result
PASS

## Checks
- Checked requirements

## Verification
- Verified tests pass

## Summary
Task completed.
`
		if string(artifact) != want {
			t.Errorf("artifact:\n%s\nwant:\n%s", string(artifact), want)
		}
		// The template selects only the internal artifact shape; the
		// caller-visible final response is identical to the custom-template run.
		if string(result.Output) != wantOutput {
			t.Errorf("caller output = %q, want %q", string(result.Output), wantOutput)
		}
	})

	// A stage run through the coordinator writes exactly that artifact through
	// the closed REVIEW selector and returns the same caller-visible output.
	t.Run("stage run writes the custom artifact and preserves caller output", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.binding.Templates.Review = customReviewStageTemplate

		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{
			reviewPassBody(wantOutput, "Task completed."),
		}, &bodies), bridge)

		result, err := stage.run(context.Background(), reviewRequest(t), ctrl)
		if err != nil {
			t.Fatalf("review stage run failed: %v", err)
		}

		want := `# Review

Operator preamble.

## Result
PASS

## Checks
- Checked requirements

## Verification
- Verified tests pass

## Summary
Task completed.

Operator footer.
`
		if len(ctrl.writes) != 1 || ctrl.writes[0] != edgeservice.SingleRequestArtifactReview {
			t.Fatalf("artifact writes = %v, want exactly one REVIEW write", ctrl.writes)
		}
		if string(ctrl.artifact) != want {
			t.Errorf("written artifact:\n%s\nwant:\n%s", string(ctrl.artifact), want)
		}
		if string(result.Output) != wantOutput {
			t.Errorf("caller output = %q, want %q", string(result.Output), wantOutput)
		}
		// The internal template must not leak into the caller-visible output.
		if strings.Contains(string(result.Output), "Operator preamble") || strings.Contains(string(result.Output), "## Checks") {
			t.Errorf("caller output leaked internal artifact text: %q", string(result.Output))
		}
	})

	t.Run("an invalid effective template fails the render closed", func(t *testing.T) {
		if _, _, err := renderSingleRequestReview(strings.Replace(customReviewStageTemplate, "PASS", "NOTPASS", 1), decision, 4096); !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err = %v, want errSingleRequestReviewStage", err)
		}
	})
}

func expectedSingleRequestReviewBodyAuthority(isResumed bool) map[string]any {
	var raw string
	if !isResumed {
		raw = `{
  "model": "gemini-3.6-flash",
  "reasoning_effort": "high",
  "temperature": 0.2,
  "tool_choice": "auto",
  "parallel_tool_calls": false,
  "stream": false,
  "messages": [
    {
      "role": "system",
      "content": "Review the task, plan, completed work, and verification evidence. Return exactly one JSON object with decision=pass, non-empty output, non-empty checks, non-empty verification, and non-empty summary when approved. Otherwise make exactly one approved workspace tool call to inspect or repair, with no text content. Every relative_path argument must be canonical and workspace-relative: use README.md, never ./README.md, an absolute path, or a parent traversal. After a tool result with error_code=invalid_request, correct the arguments and make exactly one valid tool call; do not pass. After a tool result with error_code=not_found, do not pass or inspect again; make one repair tool call."
    },
    {
      "role": "user",
      "content": "Task:\nupdate file\n\nPLAN:\n# Plan\n\nWrite result.txt.\n\n\nWORK COMPLETION:\nUpdated result.txt.\n\nWORK VERIFICATION:\nverify passed"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "workspace_read",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_list",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_write",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path", "content"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            },
            "content": {
              "type": "string"
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_delete",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_command",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["command_id"],
          "properties": {
            "command_id": {
              "type": "string",
              "enum": ["verify"]
            },
            "environment": {
              "type": "object",
              "additionalProperties": false,
              "properties": {
                "SAFE": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  ]
}`
	} else {
		raw = `{
  "model": "gemini-3.6-flash",
  "reasoning_effort": "high",
  "temperature": 0.2,
  "tool_choice": "auto",
  "parallel_tool_calls": false,
  "stream": false,
  "messages": [
    {
      "role": "system",
      "content": "Review the task, plan, completed work, and verification evidence. Return exactly one JSON object with decision=pass, non-empty output, non-empty checks, non-empty verification, and non-empty summary when approved. Otherwise make exactly one approved workspace tool call to inspect or repair, with no text content. Every relative_path argument must be canonical and workspace-relative: use README.md, never ./README.md, an absolute path, or a parent traversal. After a tool result with error_code=invalid_request, correct the arguments and make exactly one valid tool call; do not pass. After a tool result with error_code=not_found, do not pass or inspect again; make one repair tool call."
    },
    {
      "role": "user",
      "content": "Task:\nupdate file\n\nPLAN:\n# Plan\n\nWrite result.txt.\n\n\nWORK COMPLETION:\nUpdated result.txt.\n\nWORK VERIFICATION:\nverify passed"
    },
    {
      "role": "assistant",
      "content": "",
      "tool_calls": [
        {
          "id": "inspect-exact",
          "type": "function",
          "function": {
            "name": "workspace_read",
            "arguments": "{\"relative_path\":\"result.txt\"}"
          },
          "extra_content": {
            "google": {
              "thought_signature": "provider-private-tool-signature"
            }
          }
        }
      ]
    },
    {
      "role": "tool",
      "tool_call_id": "inspect-exact",
      "tool_name": "workspace_read",
      "content": "{\"content\":\"\",\"entries\":null,\"error_code\":\"\",\"exit_code\":0,\"status\":\"success\",\"stderr\":\"\",\"stdout\":\"inspection complete\",\"truncated\":false}"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "workspace_read",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_list",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_write",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path", "content"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            },
            "content": {
              "type": "string"
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_delete",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["relative_path"],
          "properties": {
            "relative_path": {
              "type": "string",
              "minLength": 1,
              "description": "Canonical path relative to the workspace root. Never start with /, ./, or ../; use README.md rather than ./README.md."
            }
          }
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "workspace_command",
        "description": "Approved IOP workspace operation.",
        "parameters": {
          "type": "object",
          "additionalProperties": false,
          "required": ["command_id"],
          "properties": {
            "command_id": {
              "type": "string",
              "enum": ["verify"]
            },
            "environment": {
              "type": "object",
              "additionalProperties": false,
              "properties": {
                "SAFE": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  ]
}`
	}
	var res map[string]any
	_ = json.Unmarshal([]byte(raw), &res)
	for _, rawTool := range res["tools"].([]any) {
		tool := rawTool.(map[string]any)
		function := tool["function"].(map[string]any)
		function["strict"] = true
	}
	encodedFormat, _ := json.Marshal(singleRequestReviewResponseFormat())
	var expectedFormat any
	_ = json.Unmarshal(encodedFormat, &expectedFormat)
	res["response_format"] = expectedFormat
	return res
}

func assertSingleRequestReviewDispatchAuthority(t *testing.T, got edgeservice.ProviderPoolDispatchRequest, dispatch *edgeservice.SingleRequestStageDispatchBinding) {
	t.Helper()
	expectedRun := edgeservice.SubmitRunRequest{
		NodeRef:          "node",
		ModelGroupKey:    dispatch.ModelGroupKey,
		ProviderID:       dispatch.ProviderID,
		UsageAttribution: "principal",
		SessionID:        "review-session",
		TimeoutSec:       dispatch.TimeoutSec,
		MaxQueue:         dispatch.MaxQueue,
		QueueTimeoutMS:   dispatch.QueueTimeoutMS,
		ProviderPool:     true,
	}
	if !reflect.DeepEqual(got.Run, expectedRun) {
		t.Fatalf("Run mismatch:\n got: %#v\nwant: %#v", got.Run, expectedRun)
	}

	if got.Tunnel.BuildBody == nil {
		t.Fatal("missing BuildBody in dispatch Tunnel request")
	}
	gotTunnel := got.Tunnel
	gotTunnel.BuildBody = nil
	expectedTunnel := edgeservice.SubmitProviderTunnelRequest{
		CredentialBinding: dispatch.CredentialBindingSnapshot(),
		NodeRef:           "node",
		ModelGroupKey:     dispatch.ModelGroupKey,
		ProviderID:        dispatch.ProviderID,
		UsageAttribution:  "principal",
		Adapter:           "openai_compat",
		Target:            dispatch.UpstreamModel,
		SessionID:         "review-session",
		Method:            "POST",
		Path:              "/v1/chat/completions",
		Operation:         "",
		Stream:            false,
		TimeoutSec:        dispatch.TimeoutSec,
		MaxQueue:          dispatch.MaxQueue,
		QueueTimeoutMS:    dispatch.QueueTimeoutMS,
		ProviderPool:      true,
	}
	if !reflect.DeepEqual(gotTunnel, expectedTunnel) {
		t.Fatalf("Tunnel mismatch:\n got: %#v\nwant: %#v", gotTunnel, expectedTunnel)
	}

	if got.AcceptCandidate == nil {
		t.Fatal("missing AcceptCandidate predicate")
	}
	profile, err := config.ResolveProtocolProfile("gemini", "", config.BuiltInProtocolProfileCatalog())
	if err != nil {
		t.Fatal(err)
	}
	accepted := edgeservice.ProviderPoolCandidate{ProviderID: dispatch.ProviderID, ExecutionPath: string(edgeservice.ProviderPoolPathTunnel), ProtocolProfile: &profile}
	if !got.AcceptCandidate(accepted) {
		t.Fatalf("AcceptCandidate rejected matching provider %q", dispatch.ProviderID)
	}
	accepted.ProviderID = "rejected-provider"
	if got.AcceptCandidate(accepted) {
		t.Fatalf("AcceptCandidate accepted non-matching provider")
	}
}

func TestSingleRequestReviewStageExactBodyAuthority(t *testing.T) {
	bridge := newSingleRequestWorkToolBridge()
	ctrl := newReviewController(t, bridge)
	ctrl.binding.Workspace.OperationIDs = []string{"read", "list", "write", "delete", "command"}
	var bodies [][]byte
	var capturedDispatches []edgeservice.ProviderPoolDispatchRequest
	responses := [][]byte{
		reviewToolBody("inspect-exact", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
		reviewPassBody("Approved output.", "Deep authority verified."),
	}
	var mu sync.Mutex
	index := 0
	provider := newSingleRequestProviderStage(&mockService{submit: func(_ context.Context, request edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
		body, err := request.Tunnel.BuildBody("gemini-3.6-flash")
		if err != nil {
			return nil, err
		}
		mu.Lock()
		bodies = append(bodies, body)
		capturedDispatches = append(capturedDispatches, request)
		current := index
		index++
		mu.Unlock()
		if current >= len(responses) {
			return nil, errors.New("unexpected provider call")
		}
		return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[current])}, DispatchInfo: matchingDispatch()}, nil
	}})
	stage := newSingleRequestReviewStage(provider, bridge)
	result, err := stage.run(context.Background(), reviewRequest(t), ctrl)
	if err != nil {
		t.Fatal(err)
	}
	if string(result.Output) != "Approved output." || result.Summary != "Deep authority verified." {
		t.Fatalf("unexpected result: %+v", result)
	}
	if len(bodies) != 2 || len(capturedDispatches) != 2 {
		t.Fatalf("expected 2 bodies and dispatches, got %d bodies, %d dispatches", len(bodies), len(capturedDispatches))
	}

	dispatch := ctrl.binding.Review.Dispatch
	for _, req := range capturedDispatches {
		assertSingleRequestReviewDispatchAuthority(t, req, dispatch)
	}

	for i, raw := range bodies {
		isResumed := i == 1
		var payload map[string]any
		if err := json.Unmarshal(raw, &payload); err != nil {
			t.Fatalf("failed to unmarshal body %d: %v", i, err)
		}
		messages, ok := payload["messages"].([]any)
		if !ok || len(messages) < 2 || !strings.Contains(string(raw), "REVIEW HANDOFF") || !strings.Contains(string(raw), "Worker Item Status") {
			t.Fatalf("review body did not preserve the artifact-only handoff (isResumed=%v): %s", isResumed, raw)
		}
	}
}

type failingEnvelopeController struct {
	*reviewController
	failOnStage edgeservice.SingleRequestState
}

func (c *failingEnvelopeController) SubmitEnvelope(env edgeservice.SingleRequestEnvelope) error {
	if env.Stage == c.failOnStage {
		return errors.New("targeted envelope error")
	}
	return c.reviewController.SubmitEnvelope(env)
}

func TestSingleRequestReviewStageFailureMatrix(t *testing.T) {
	t.Run("provider failure returns review stage error and leaves zero waiters", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		stage := newSingleRequestReviewStage(newSingleRequestProviderStage(&mockService{
			submit: func(_ context.Context, _ edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
				return nil, errors.New("provider failure")
			},
		}), bridge)
		_, err := stage.run(context.Background(), reviewRequest(t), ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if len(ctrl.artifact) > 0 || len(ctrl.writes) > 0 {
			t.Fatalf("artifact or writes modified on provider error: artifact=%q writes=%v", ctrl.artifact, ctrl.writes)
		}
		if bridge.pendingCount() != 0 {
			t.Fatalf("pendingCount=%d, want 0", bridge.pendingCount())
		}
	})

	t.Run("envelope failure on initial envelope returns error and cleans up", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.envelopeErr = errors.New("envelope submission failed")
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewPassBody("output", "summary")}, &bodies), bridge)
		_, err := stage.run(context.Background(), reviewRequest(t), ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if bridge.pendingCount() != 0 {
			t.Fatalf("pendingCount=%d, want 0", bridge.pendingCount())
		}
	})

	t.Run("envelope failure on tool call envelope unregisters waiter", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		failCtrl := &failingEnvelopeController{reviewController: ctrl, failOnStage: edgeservice.SingleRequestStateInternalTool}
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{reviewToolBody("inspect-fail-env", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`)}, &bodies), bridge)
		_, err := stage.run(context.Background(), reviewRequest(t), failCtrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if bridge.pendingCount() != 0 {
			t.Fatalf("pendingCount=%d, want 0", bridge.pendingCount())
		}
	})
}

func TestSingleRequestReviewStageContinuationCorrelation(t *testing.T) {
	t.Run("mismatch and concurrent duplicate delivery", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.autoContinue = false
		var bodies [][]byte
		responses := [][]byte{
			reviewToolBody("corr-tool-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
			reviewPassBody("Approved correlated output.", "Correlation summary."),
		}
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, responses, &bodies), bridge)

		ctx := context.Background()
		done := make(chan error, 1)
		go func() {
			_, err := stage.run(ctx, reviewRequest(t), ctrl)
			done <- err
		}()

		deadline := time.After(2 * time.Second)
		for bridge.pendingCount() != 1 {
			select {
			case <-deadline:
				t.Fatal("tool waiter was not registered in time")
			case <-time.After(time.Millisecond):
			}
		}

		errWrongReq := bridge.ContinueInternalTool(ctx, edgeservice.InternalWorkspaceToolResult{
			RequestID: "wrong-request-id", StageID: singleRequestReviewStageID, ToolCallID: "corr-tool-1", Status: "success", Stdout: []byte("wrong"),
		})
		if errWrongReq == nil || bridge.pendingCount() != 1 {
			t.Fatalf("wrong request ID err=%v pendingCount=%d", errWrongReq, bridge.pendingCount())
		}

		errWrongStage := bridge.ContinueInternalTool(ctx, edgeservice.InternalWorkspaceToolResult{
			RequestID: "request-review", StageID: "wrong-stage", ToolCallID: "corr-tool-1", Status: "success", Stdout: []byte("wrong"),
		})
		if errWrongStage == nil || bridge.pendingCount() != 1 {
			t.Fatalf("wrong stage ID err=%v pendingCount=%d", errWrongStage, bridge.pendingCount())
		}

		errWrongTool := bridge.ContinueInternalTool(ctx, edgeservice.InternalWorkspaceToolResult{
			RequestID: "request-review", StageID: singleRequestReviewStageID, ToolCallID: "wrong-tool-id", Status: "success", Stdout: []byte("wrong"),
		})
		if errWrongTool == nil || bridge.pendingCount() != 1 {
			t.Fatalf("wrong tool call ID err=%v pendingCount=%d", errWrongTool, bridge.pendingCount())
		}

		validResult := edgeservice.InternalWorkspaceToolResult{
			RequestID: "request-review", StageID: singleRequestReviewStageID, ToolCallID: "corr-tool-1", Status: "success", Stdout: []byte("correlated inspection result"),
		}

		var wg sync.WaitGroup
		errs := make(chan error, 2)
		wg.Add(2)
		go func() {
			defer wg.Done()
			errs <- bridge.ContinueInternalTool(ctx, validResult)
		}()
		go func() {
			defer wg.Done()
			errs <- bridge.ContinueInternalTool(ctx, validResult)
		}()
		wg.Wait()
		close(errs)

		var errList []error
		for err := range errs {
			errList = append(errList, err)
		}
		if len(errList) != 2 {
			t.Fatalf("expected 2 errors from race, got %d", len(errList))
		}
		if (errList[0] == nil && errList[1] == nil) || (errList[0] != nil && errList[1] != nil) {
			t.Fatalf("expected exactly one success and one failure, got err0=%v err1=%v", errList[0], errList[1])
		}

		errStale := bridge.ContinueInternalTool(ctx, validResult)
		if errStale == nil {
			t.Fatalf("expected error on duplicate continuation delivery, got nil")
		}

		select {
		case err := <-done:
			if err != nil {
				t.Fatalf("stage.run returned unexpected error: %v", err)
			}
		case <-time.After(2 * time.Second):
			t.Fatal("stage.run timed out")
		}

		if bridge.pendingCount() != 0 {
			t.Fatalf("final pendingCount=%d, want 0", bridge.pendingCount())
		}
	})

	t.Run("post-cancel stale delivery rejected", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.autoContinue = false
		var bodies [][]byte
		responses := [][]byte{
			reviewToolBody("stale-tool-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
		}
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, responses, &bodies), bridge)

		ctx, cancel := context.WithCancel(context.Background())
		done := make(chan error, 1)
		go func() {
			_, err := stage.run(ctx, reviewRequest(t), ctrl)
			done <- err
		}()

		deadline := time.After(2 * time.Second)
		for bridge.pendingCount() != 1 {
			select {
			case <-deadline:
				t.Fatal("tool waiter was not registered in time")
			case <-time.After(time.Millisecond):
			}
		}

		cancel()

		select {
		case err := <-done:
			if !errors.Is(err, errSingleRequestReviewStage) {
				t.Fatalf("stage.run err=%v, want errSingleRequestReviewStage", err)
			}
		case <-time.After(2 * time.Second):
			t.Fatal("stage.run timed out on cancellation")
		}

		if bridge.pendingCount() != 0 {
			t.Fatalf("pendingCount=%d after cancellation, want 0", bridge.pendingCount())
		}

		staleResult := edgeservice.InternalWorkspaceToolResult{
			RequestID: "request-review", StageID: singleRequestReviewStageID, ToolCallID: "stale-tool-1", Status: "success", Stdout: []byte("stale result"),
		}
		if err := bridge.ContinueInternalTool(context.Background(), staleResult); err == nil {
			t.Fatal("expected error on post-cancel stale delivery, got nil")
		}
	})
}

func TestSingleRequestReviewStageCommandVerificationAfterMutation(t *testing.T) {
	t.Run("not found then command repair then command verification passes", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		ctrl.toolResult = func(call *edgeservice.InternalWorkspaceToolCall) edgeservice.InternalWorkspaceToolResult {
			result := edgeservice.InternalWorkspaceToolResult{RequestID: call.RequestID, StageID: call.StageID, ToolCallID: call.ToolCallID, Status: "success", Stdout: []byte("verified")}
			if call.ToolCallID == "missing-1" {
				result.Status = "error"
				result.ErrorCode = "not_found"
				result.Stdout = nil
			}
			if call.ToolCallID == "repair-command-1" {
				result.Stdout = []byte("repaired")
			}
			return result
		}
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{
			reviewToolBody("missing-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
			reviewToolBody("repair-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewToolBody("verify-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewPassBody("Command repaired and verified.", "Missing artifact was repaired by command and verified."),
		}, &bodies), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 4
		result, err := stage.run(context.Background(), req, ctrl)
		if err != nil || string(result.Output) != "Command repaired and verified." {
			t.Fatalf("stage.run result=%+v err=%v", result, err)
		}
		if bridge.pendingCount() != 0 || len(ctrl.writes) != 0 {
			t.Fatalf("pending=%d review writes=%d", bridge.pendingCount(), len(ctrl.writes))
		}
		if len(bodies) != 4 || !containsAll(string(bodies[1]), "not_found", "missing-1") || !containsAll(string(bodies[2]), "repair-command-1") || !containsAll(string(bodies[3]), "verify-command-1", "repaired") {
			t.Fatalf("provider continuation bodies=%q", bodies)
		}
		wantChoices := []string{"auto", "required", "auto", "auto"}
		for i, body := range bodies {
			var decoded map[string]any
			if err := json.Unmarshal(body, &decoded); err != nil || decoded["tool_choice"] != wantChoices[i] {
				t.Fatalf("body %d tool_choice=%v error=%v, want %s", i, decoded["tool_choice"], err, wantChoices[i])
			}
		}
	})

	t.Run("write then successful command passes without later inspection", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		statesAtDispatch := make([]edgeservice.SingleRequestState, 0, 4)
		stage := newSingleRequestReviewStage(newSingleRequestProviderStage(&mockService{submit: func(_ context.Context, request edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			body, err := request.Tunnel.BuildBody("gemini-3.6-flash")
			if err != nil {
				return nil, err
			}
			bodies = append(bodies, body)
			statesAtDispatch = append(statesAtDispatch, ctrl.State())
			responses := [][]byte{
				reviewToolBody("repair-write-1", edgeservice.InternalWorkspaceToolWrite, `{"relative_path":"result.txt","content":"fixed"}`),
				reviewToolBody("verify-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
				reviewPassBody("Repaired and verified.", "Command verified the mutation."),
			}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[len(bodies)-1])}, DispatchInfo: matchingDispatch()}, nil
		}}), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 4
		result, err := stage.run(context.Background(), req, ctrl)
		if err != nil {
			t.Fatalf("stage.run err=%v", err)
		}
		if string(result.Output) != "Repaired and verified." {
			t.Fatalf("output=%q, want 'Repaired and verified.'", result.Output)
		}
		if bridge.pendingCount() != 0 || len(ctrl.writes) != 0 {
			t.Fatalf("pending=%d writes=%d", bridge.pendingCount(), len(ctrl.writes))
		}
		if len(ctrl.envelopes) != 7 {
			t.Fatalf("envelopes=%+v", ctrl.envelopes)
		}
		wantStates := []edgeservice.SingleRequestState{
			edgeservice.SingleRequestStateReviewing,
			edgeservice.SingleRequestStateRepairing,
			edgeservice.SingleRequestStateInternalTool,
			edgeservice.SingleRequestStateRepairing,
			edgeservice.SingleRequestStateInternalTool,
			edgeservice.SingleRequestStateRepairing,
			edgeservice.SingleRequestStateFinalizing,
		}
		for i, s := range wantStates {
			if ctrl.envelopes[i].Stage != s {
				t.Fatalf("envelopes[%d]=%v, want %v", i, ctrl.envelopes[i].Stage, s)
			}
		}
		if len(statesAtDispatch) != 3 || statesAtDispatch[0] != edgeservice.SingleRequestStateReviewing || statesAtDispatch[1] != edgeservice.SingleRequestStateRepairing || statesAtDispatch[2] != edgeservice.SingleRequestStateRepairing {
			t.Fatalf("dispatch states=%v", statesAtDispatch)
		}
	})

	t.Run("command first without prior mutation cannot pass", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{
			reviewToolBody("command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewPassBody("Should not pass.", "Summary."),
		}, &bodies), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 2
		_, err := stage.run(context.Background(), req, ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if len(ctrl.writes) != 0 {
			t.Fatalf("review artifact writes=%d, want 0", len(ctrl.writes))
		}
		if bridge.pendingCount() != 0 {
			t.Fatalf("pending=%d, want 0", bridge.pendingCount())
		}
		if len(bodies) != 2 {
			t.Fatalf("bodies=%d, want 2", len(bodies))
		}
	})

	t.Run("command after inspection without mutation cannot pass", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		stage := newSingleRequestReviewStage(scriptedReviewProvider(t, ctrl, [][]byte{
			reviewToolBody("inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
			reviewToolBody("command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewPassBody("Should not pass.", "Summary."),
		}, &bodies), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 3
		_, err := stage.run(context.Background(), req, ctrl)
		if !errors.Is(err, errSingleRequestReviewStage) {
			t.Fatalf("err=%v, want errSingleRequestReviewStage", err)
		}
		if bridge.pendingCount() != 0 {
			t.Fatalf("pending=%d, want 0", bridge.pendingCount())
		}
	})

	t.Run("write then command then inspection still passes", func(t *testing.T) {
		bridge := newSingleRequestWorkToolBridge()
		ctrl := newReviewController(t, bridge)
		var bodies [][]byte
		stage := newSingleRequestReviewStage(newSingleRequestProviderStage(&mockService{submit: func(_ context.Context, request edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			body, err := request.Tunnel.BuildBody("gemini-3.6-flash")
			if err != nil {
				return nil, err
			}
			bodies = append(bodies, body)
			responses := [][]byte{
				reviewToolBody("repair-write-1", edgeservice.InternalWorkspaceToolWrite, `{"relative_path":"result.txt","content":"fixed"}`),
				reviewToolBody("verify-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
				reviewToolBody("inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
				reviewPassBody("Repaired and verified.", "Command and inspection verified the mutation."),
			}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[len(bodies)-1])}, DispatchInfo: matchingDispatch()}, nil
		}}), bridge)
		req := reviewRequest(t)
		req.Limits.MaxToolIterations = 4
		result, err := stage.run(context.Background(), req, ctrl)
		if err != nil {
			t.Fatalf("stage.run err=%v", err)
		}
		if string(result.Output) != "Repaired and verified." {
			t.Fatalf("output=%q", result.Output)
		}
		if bridge.pendingCount() != 0 || len(ctrl.writes) != 0 {
			t.Fatalf("pending=%d writes=%d", bridge.pendingCount(), len(ctrl.writes))
		}
	})
}

func TestSingleRequestReviewStageCommandVerificationCoordinator(t *testing.T) {
	t.Run("write then command passes through coordinator", func(t *testing.T) {
		var providerCalls atomic.Int32
		responses := [][]byte{
			reviewToolBody("repair-1", edgeservice.InternalWorkspaceToolWrite, `{"relative_path":"result.txt","content":"fixed"}`),
			reviewToolBody("verify-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewPassBody("Coordinator verified.", "Command verified the repair."),
		}
		provider := &mockService{submit: func(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			if _, err := req.Tunnel.BuildBody("gemini-3.6-flash"); err != nil {
				return nil, err
			}
			index := int(providerCalls.Add(1)) - 1
			if index >= len(responses) {
				return nil, errors.New("unexpected provider call")
			}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[index])}, DispatchInfo: matchingDispatch()}, nil
		}}
		harness := newReviewCoordinatorHarness(t, provider, nil)
		harness.node.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
			response := &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}
			if req.GetOperation() == iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND {
				response.Stdout = []byte("verified")
			}
			return response
		}

		execution, err := harness.service.StartSingleRequest(context.Background(), reviewServiceRequest(harness.binding))
		if err != nil {
			t.Fatalf("StartSingleRequest: %v", err)
		}
		waitWorkFinalizing(t, execution)
		if err := execution.AcknowledgeTerminal(true); err != nil {
			t.Fatalf("AcknowledgeTerminal: %v", err)
		}
		result, waitErr := waitWorkExecution(t, execution)
		if waitErr != nil || result.Output != "Coordinator verified." {
			t.Fatalf("Wait=(%q,%v)", result.Output, waitErr)
		}
		var outcome reviewStageExecutionOutcome
		select {
		case outcome = <-harness.executor.outcomes:
		case <-time.After(3 * time.Second):
			t.Fatal("timed out waiting for executor outcome")
		}
		if outcome.err != nil || outcome.result == nil || outcome.result.Summary != "Command verified the repair." {
			t.Fatalf("executor outcome=%+v", outcome)
		}
		if providerCalls.Load() != 3 || harness.node.toolCount.Load() != 2 || harness.executor.continueCount.Load() != 2 || harness.executor.reviewWriteCount.Load() != 1 || harness.executor.finalizingCount.Load() != 1 || harness.node.cleanupCount.Load() != 1 || harness.bridge.pendingCount() != 0 {
			t.Fatalf("provider=%d tool=%d continuations=%d reviewWrites=%d finalizing=%d cleanup=%d pending=%d", providerCalls.Load(), harness.node.toolCount.Load(), harness.executor.continueCount.Load(), harness.executor.reviewWriteCount.Load(), harness.executor.finalizingCount.Load(), harness.node.cleanupCount.Load(), harness.bridge.pendingCount())
		}
	})

	t.Run("command first without mutation fails through coordinator", func(t *testing.T) {
		var providerCalls atomic.Int32
		responses := [][]byte{
			reviewToolBody("command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
			reviewPassBody("Should not pass.", "Summary."),
		}
		provider := &mockService{submit: func(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			if _, err := req.Tunnel.BuildBody("gemini-3.6-flash"); err != nil {
				return nil, err
			}
			index := int(providerCalls.Add(1)) - 1
			if index >= len(responses) {
				return nil, errors.New("unexpected provider call")
			}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[index])}, DispatchInfo: matchingDispatch()}, nil
		}}
		harness := newReviewCoordinatorHarness(t, provider, nil)
		harness.node.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
			return &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Stdout: []byte("verified")}
		}

		execution, err := harness.service.StartSingleRequest(context.Background(), reviewServiceRequest(harness.binding))
		if err != nil {
			t.Fatalf("StartSingleRequest: %v", err)
		}
		_, waitErr := waitWorkExecution(t, execution)
		var outcome reviewStageExecutionOutcome
		select {
		case outcome = <-harness.executor.outcomes:
		case <-time.After(3 * time.Second):
			t.Fatal("timed out waiting for executor outcome")
		}
		if waitErr == nil || !errors.Is(outcome.err, errSingleRequestReviewStage) {
			t.Fatalf("Wait=(%q,%v) outcome=%+v", waitErr, waitErr, outcome)
		}
		if providerCalls.Load() != 2 || harness.executor.reviewWriteCount.Load() != 1 || harness.executor.finalizingCount.Load() != 0 || harness.node.cleanupCount.Load() != 1 || harness.bridge.pendingCount() != 0 {
			t.Fatalf("provider=%d reviewWrites=%d finalizing=%d cleanup=%d pending=%d", providerCalls.Load(), harness.executor.reviewWriteCount.Load(), harness.executor.finalizingCount.Load(), harness.node.cleanupCount.Load(), harness.bridge.pendingCount())
		}
	})
}

func TestSingleRequestReviewStageCoordinatorToolFailure(t *testing.T) {
	t.Run("typed Node tool failure causes stage fail-closed with zero leak", func(t *testing.T) {
		var providerCalls atomic.Int32
		responses := [][]byte{
			reviewToolBody("repair-fail-1", edgeservice.InternalWorkspaceToolWrite, `{"relative_path":"result.txt","content":"bad"}`),
			reviewPassBody("Approved output.", "Summary."),
		}
		provider := &mockService{submit: func(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
			if _, err := req.Tunnel.BuildBody("gemini-3.6-flash"); err != nil {
				return nil, err
			}
			index := int(providerCalls.Add(1)) - 1
			if index >= len(responses) {
				return nil, errors.New("unexpected provider call")
			}
			return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[index])}, DispatchInfo: matchingDispatch()}, nil
		}}
		harness := newReviewCoordinatorHarness(t, provider, nil)
		harness.node.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
			return &iop.WorkspaceToolResponse{
				RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
				Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR,
				ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
				Error:     "node execution failed: permission denied",
			}
		}

		execution, err := harness.service.StartSingleRequest(context.Background(), reviewServiceRequest(harness.binding))
		if err != nil {
			t.Fatalf("StartSingleRequest: %v", err)
		}

		type waitResult struct {
			result edgeservice.SingleRequestResult
			err    error
		}
		done := make(chan waitResult, 1)
		go func() {
			res, err := execution.Wait()
			done <- waitResult{result: res, err: err}
		}()

		var waitRes waitResult
		select {
		case waitRes = <-done:
		case <-time.After(3 * time.Second):
			t.Fatal("timed out waiting for execution")
		}

		if !errors.Is(waitRes.err, edgeservice.ErrSingleRequestInternalToolFailed) || strings.Contains(waitRes.err.Error(), "permission denied") {
			t.Fatalf("wait err=%v, want ErrSingleRequestInternalToolFailed without raw leak", waitRes.err)
		}

		var outcome reviewStageExecutionOutcome
		select {
		case outcome = <-harness.executor.outcomes:
		case <-time.After(3 * time.Second):
			t.Fatal("timed out waiting for executor outcome")
		}

		if !errors.Is(outcome.err, errSingleRequestReviewStage) {
			t.Fatalf("executor outcome err=%v, want errSingleRequestReviewStage", outcome.err)
		}

		if providerCalls.Load() != 1 {
			t.Fatalf("providerCalls=%d, want 1", providerCalls.Load())
		}
		if harness.node.toolCount.Load() != 1 {
			t.Fatalf("toolCount=%d, want 1", harness.node.toolCount.Load())
		}
		if harness.executor.continueCount.Load() != 0 {
			t.Fatalf("continueCount=%d, want 0", harness.executor.continueCount.Load())
		}
		if harness.node.cleanupCount.Load() != 1 {
			t.Fatalf("cleanupCount=%d, want 1", harness.node.cleanupCount.Load())
		}
		if harness.bridge.pendingCount() != 0 {
			t.Fatalf("pendingCount=%d, want 0", harness.bridge.pendingCount())
		}
		if harness.executor.reviewWriteCount.Load() != 1 || harness.executor.finalizingCount.Load() != 0 || execution.State() != edgeservice.SingleRequestStateFailed || waitRes.result.Output != "" {
			t.Fatalf("review/finalizing leak: writes=%d finalizing=%d state=%s result=%q", harness.executor.reviewWriteCount.Load(), harness.executor.finalizingCount.Load(), execution.State(), waitRes.result.Output)
		}
	})
}

func TestSingleRequestReviewStageCoordinatorRepairsMissingArtifact(t *testing.T) {
	responses := [][]byte{
		reviewToolBody("missing-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
		reviewToolBody("repair-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
		reviewToolBody("verify-command-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify","environment":{"SAFE":"1"}}`),
		reviewPassBody("Repaired output.", "Missing artifact was repaired and verified."),
	}
	var providerCalls atomic.Int32
	var bodiesMu sync.Mutex
	var bodies [][]byte
	provider := &mockService{submit: func(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
		body, err := req.Tunnel.BuildBody("gemini-3.6-flash")
		if err != nil {
			return nil, err
		}
		index := int(providerCalls.Add(1)) - 1
		if index >= len(responses) {
			return nil, errors.New("unexpected provider call")
		}
		bodiesMu.Lock()
		bodies = append(bodies, append([]byte(nil), body...))
		bodiesMu.Unlock()
		return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[index])}, DispatchInfo: matchingDispatch()}, nil
	}}
	harness := newReviewCoordinatorHarness(t, provider, nil)
	harness.node.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
		response := &iop.WorkspaceToolResponse{
			RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
			Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
		}
		switch req.GetToolCallId() {
		case "missing-1":
			response.Status = iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR
			response.ErrorCode = iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_NOT_FOUND
			response.Error = "workspace entry not found"
		case "repair-command-1":
			response.Stdout = []byte("fixed by approved command")
		case "verify-command-1":
			response.Stdout = []byte("fixed")
		}
		return response
	}

	execution, err := harness.service.StartSingleRequest(context.Background(), reviewServiceRequest(harness.binding))
	if err != nil {
		t.Fatalf("StartSingleRequest: %v", err)
	}
	waitWorkFinalizing(t, execution)
	if err := execution.AcknowledgeTerminal(true); err != nil {
		t.Fatalf("AcknowledgeTerminal: %v", err)
	}
	result, waitErr := waitWorkExecution(t, execution)
	if waitErr != nil || result.Output != "Repaired output." {
		t.Fatalf("Wait=(%q,%v)", result.Output, waitErr)
	}
	var outcome reviewStageExecutionOutcome
	select {
	case outcome = <-harness.executor.outcomes:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for executor outcome")
	}
	if outcome.err != nil || outcome.result == nil || outcome.result.Summary != "Missing artifact was repaired and verified." {
		t.Fatalf("executor outcome=%+v", outcome)
	}

	bodiesMu.Lock()
	captured := append([][]byte(nil), bodies...)
	bodiesMu.Unlock()
	if len(captured) != 4 || !containsAll(string(captured[1]), "not_found", "missing-1") || !containsAll(string(captured[2]), "repair-command-1") || !containsAll(string(captured[3]), "verify-command-1", "fixed") {
		t.Fatalf("provider continuation bodies=%q", captured)
	}
	wantChoices := []string{"auto", "required", "auto", "auto"}
	for i, body := range captured {
		var decoded map[string]any
		if err := json.Unmarshal(body, &decoded); err != nil || decoded["tool_choice"] != wantChoices[i] {
			t.Fatalf("body %d tool_choice=%v error=%v, want %s", i, decoded["tool_choice"], err, wantChoices[i])
		}
	}
	if providerCalls.Load() != 4 || harness.node.toolCount.Load() != 3 || harness.executor.continueCount.Load() != 3 || harness.executor.reviewWriteCount.Load() != 1 || harness.executor.finalizingCount.Load() != 1 || harness.node.cleanupCount.Load() != 1 || harness.bridge.pendingCount() != 0 {
		t.Fatalf("provider=%d tool=%d continuations=%d reviewWrites=%d finalizing=%d cleanup=%d pending=%d", providerCalls.Load(), harness.node.toolCount.Load(), harness.executor.continueCount.Load(), harness.executor.reviewWriteCount.Load(), harness.executor.finalizingCount.Load(), harness.node.cleanupCount.Load(), harness.bridge.pendingCount())
	}
}

func TestSingleRequestReviewStageCoordinatorRejectsMissingArtifactWithoutRepair(t *testing.T) {
	tests := []struct {
		name   string
		second []byte
	}{
		{name: "pass", second: reviewPassBody("Unrepaired output.", "Missing artifact was ignored.")},
		{name: "inspection", second: reviewToolBody("inspect-again", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`)},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			responses := [][]byte{
				reviewToolBody("missing-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
				test.second,
			}
			var providerCalls atomic.Int32
			provider := &mockService{submit: func(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
				body, err := req.Tunnel.BuildBody("gemini-3.6-flash")
				if err != nil {
					return nil, err
				}
				index := int(providerCalls.Add(1)) - 1
				if index >= len(responses) {
					return nil, errors.New("unexpected provider call")
				}
				var decoded map[string]any
				wantChoice := "auto"
				if index == 1 {
					wantChoice = "required"
				}
				if json.Unmarshal(body, &decoded) != nil || decoded["tool_choice"] != wantChoice {
					return nil, errors.New("unexpected review tool choice")
				}
				return &edgeservice.ProviderPoolDispatchResult{Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(responses[index])}, DispatchInfo: matchingDispatch()}, nil
			}}
			harness := newReviewCoordinatorHarness(t, provider, nil)
			harness.node.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
				return &iop.WorkspaceToolResponse{
					RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
					Status: iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_NOT_FOUND, Error: "workspace entry not found",
				}
			}

			execution, err := harness.service.StartSingleRequest(context.Background(), reviewServiceRequest(harness.binding))
			if err != nil {
				t.Fatalf("StartSingleRequest: %v", err)
			}
			result, waitErr := waitWorkExecution(t, execution)
			var outcome reviewStageExecutionOutcome
			select {
			case outcome = <-harness.executor.outcomes:
			case <-time.After(3 * time.Second):
				t.Fatal("timed out waiting for executor outcome")
			}
			if waitErr == nil || !errors.Is(outcome.err, errSingleRequestReviewStage) || result.Output != "" || execution.State() != edgeservice.SingleRequestStateFailed {
				t.Fatalf("Wait=(%q,%v) outcome=%+v state=%s", result.Output, waitErr, outcome, execution.State())
			}
			if providerCalls.Load() != 2 || harness.node.toolCount.Load() != 1 || harness.executor.continueCount.Load() != 1 || harness.executor.reviewWriteCount.Load() != 1 || harness.executor.finalizingCount.Load() != 0 || harness.node.cleanupCount.Load() != 1 || harness.bridge.pendingCount() != 0 {
				t.Fatalf("provider=%d tool=%d continuations=%d reviewWrites=%d finalizing=%d cleanup=%d pending=%d", providerCalls.Load(), harness.node.toolCount.Load(), harness.executor.continueCount.Load(), harness.executor.reviewWriteCount.Load(), harness.executor.finalizingCount.Load(), harness.node.cleanupCount.Load(), harness.bridge.pendingCount())
			}
		})
	}
}
