package service

import (
	"context"
	"errors"
	"net"
	"slices"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"git.toki-labs.com/toki/proto-socket/go/packets"
	"google.golang.org/protobuf/proto"

	edgeevents "iop/apps/edge/internal/events"
	edgenode "iop/apps/edge/internal/node"
	iop "iop/proto/gen/iop"
)

func TestWorkspaceWire(t *testing.T) {
	edgeClient, nodeClient := workspaceWirePipe(t)
	registry := edgenode.NewRegistry()
	entry := &edgenode.NodeEntry{NodeID: "node-1", Client: edgeClient}
	registry.Register(entry)
	svc := New(registry, edgeevents.NewBus())
	binding := workspaceWireBinding("workspace-1", entry.NodeID, entry.ConnectionGeneration, 1000)
	binding.Limits.MaxWriteBytes = 999
	openSeen := make(chan *iop.WorkspaceOpenRequest, 1)

	toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&nodeClient.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
		openSeen <- proto.Clone(req).(*iop.WorkspaceOpenRequest)
		return &iop.WorkspaceOpenResponse{RequestId: req.GetRequestId(), WorkspaceRef: req.GetWorkspaceRef(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}, nil
	})
	toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&nodeClient.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
		return &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}, nil
	})
	toki.AddRequestListenerTyped[*iop.WorkspaceCancelRequest, *iop.WorkspaceCancelResponse](&nodeClient.Communicator, func(req *iop.WorkspaceCancelRequest) (*iop.WorkspaceCancelResponse, error) {
		return &iop.WorkspaceCancelResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_CANCELLED, Error: "workspace command cancelled"}, nil
	})
	toki.AddRequestListenerTyped[*iop.WorkspaceCleanupRequest, *iop.WorkspaceCleanupResponse](&nodeClient.Communicator, func(req *iop.WorkspaceCleanupRequest) (*iop.WorkspaceCleanupResponse, error) {
		return &iop.WorkspaceCleanupResponse{RequestId: req.GetRequestId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, CleanedArtifacts: 1}, nil
	})

	caller := &iop.WorkspaceOpenRequest{
		RequestId: "request-1", WorkspaceRef: binding.Ref,
		Operations: []iop.WorkspaceOperation{iop.WorkspaceOperation_WORKSPACE_OPERATION_DELETE},
		CommandIds: []string{"caller-command"}, MaxReadBytes: 9999, MaxWriteBytes: 9999,
		MaxOutputBytes: 9999, MaxCommandTimeoutMs: 9999,
	}
	if response, err := svc.workspaceOpen(context.Background(), binding, caller); err != nil || response.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS {
		t.Fatalf("open = %+v, %v", response, err)
	}
	gotOpen := <-openSeen
	if !proto.Equal(gotOpen, &iop.WorkspaceOpenRequest{
		RequestId: "request-1", WorkspaceRef: "workspace-1",
		Operations: []iop.WorkspaceOperation{iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND, iop.WorkspaceOperation_WORKSPACE_OPERATION_READ},
		CommandIds: []string{"test"}, MaxReadBytes: 64, MaxOutputBytes: 64, MaxCommandTimeoutMs: 1000,
	}) {
		t.Fatalf("open authority = %+v", gotOpen)
	}
	caller.Operations[0] = iop.WorkspaceOperation_WORKSPACE_OPERATION_WRITE
	caller.CommandIds[0] = "mutated"
	if !slices.Equal(gotOpen.GetOperations(), []iop.WorkspaceOperation{iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND, iop.WorkspaceOperation_WORKSPACE_OPERATION_READ}) {
		t.Fatalf("captured authority changed after caller mutation: %+v", gotOpen)
	}
	if response, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1", Operation: iop.WorkspaceOperation_WORKSPACE_OPERATION_READ}); err != nil || response.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS {
		t.Fatalf("tool = %+v, %v", response, err)
	}
	if response, err := svc.workspaceCancel(context.Background(), binding, &iop.WorkspaceCancelRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"}); err != nil || response.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED {
		t.Fatalf("cancel = %+v, %v", response, err)
	}
	if response, err := svc.workspaceCleanup(context.Background(), binding, &iop.WorkspaceCleanupRequest{RequestId: "request-1"}); err != nil || response.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS || response.GetCleanedArtifacts() != 1 {
		t.Fatalf("cleanup = %+v, %v", response, err)
	}
}

func TestWorkspaceWireStaleGenerationNeverReselects(t *testing.T) {
	oldEdge, oldNode := workspaceWirePipe(t)
	newEdge, newNode := workspaceWirePipe(t)
	registry := edgenode.NewRegistry()
	oldEntry := &edgenode.NodeEntry{NodeID: "node-1", Client: oldEdge}
	registry.Register(oldEntry)
	binding := workspaceWireBinding("workspace-1", oldEntry.NodeID, oldEntry.ConnectionGeneration, 1000)
	registry.Register(&edgenode.NodeEntry{NodeID: "node-1", Client: newEdge})
	var reachedNew atomic.Bool
	toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&newNode.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
		reachedNew.Store(true)
		return &iop.WorkspaceOpenResponse{RequestId: req.GetRequestId(), WorkspaceRef: req.GetWorkspaceRef(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}, nil
	})
	_ = oldNode
	svc := New(registry, edgeevents.NewBus())
	if _, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref}); err != errWorkspaceWireStale {
		t.Fatalf("stale dispatch error = %v, want %v", err, errWorkspaceWireStale)
	}
	if reachedNew.Load() {
		t.Fatal("stale binding must not reselect the reconnect client")
	}
}

func workspaceWirePipe(t *testing.T) (*toki.TcpClient, *toki.TcpClient) {
	t.Helper()
	edgeConn, nodeConn := net.Pipe()
	edge := toki.NewTcpClient(edgeConn, 0, 0, workspaceWireResponseParserMap())
	node := toki.NewTcpClient(nodeConn, 0, 0, workspaceWireRequestParserMap())
	t.Cleanup(func() { _ = edge.Close(); _ = node.Close() })
	return edge, node
}

func workspaceWireRequestParserMap() toki.ParserMap {
	return toki.ParserMap{
		toki.TypeNameOf(&iop.WorkspaceOpenRequest{}):     parseWorkspaceMessage[*iop.WorkspaceOpenRequest],
		toki.TypeNameOf(&iop.WorkspaceToolRequest{}):     parseWorkspaceMessage[*iop.WorkspaceToolRequest],
		toki.TypeNameOf(&iop.WorkspaceArtifactRequest{}): parseWorkspaceMessage[*iop.WorkspaceArtifactRequest],
		toki.TypeNameOf(&iop.WorkspaceCancelRequest{}):   parseWorkspaceMessage[*iop.WorkspaceCancelRequest],
		toki.TypeNameOf(&iop.WorkspaceCleanupRequest{}):  parseWorkspaceMessage[*iop.WorkspaceCleanupRequest],
	}
}

func workspaceWireResponseParserMap() toki.ParserMap {
	return toki.ParserMap{
		toki.TypeNameOf(&iop.WorkspaceOpenResponse{}):     parseWorkspaceMessage[*iop.WorkspaceOpenResponse],
		toki.TypeNameOf(&iop.WorkspaceToolResponse{}):     parseWorkspaceMessage[*iop.WorkspaceToolResponse],
		toki.TypeNameOf(&iop.WorkspaceArtifactResponse{}): parseWorkspaceMessage[*iop.WorkspaceArtifactResponse],
		toki.TypeNameOf(&iop.WorkspaceCancelResponse{}):   parseWorkspaceMessage[*iop.WorkspaceCancelResponse],
		toki.TypeNameOf(&iop.WorkspaceCleanupResponse{}):  parseWorkspaceMessage[*iop.WorkspaceCleanupResponse],
	}
}

func parseWorkspaceMessage[T proto.Message](payload []byte) (proto.Message, error) {
	var message T
	message = newWorkspaceMessage[T]()
	return message, proto.Unmarshal(payload, message)
}

func newWorkspaceMessage[T proto.Message]() T {
	var zero T
	switch any(zero).(type) {
	case *iop.WorkspaceOpenRequest:
		return any(&iop.WorkspaceOpenRequest{}).(T)
	case *iop.WorkspaceToolRequest:
		return any(&iop.WorkspaceToolRequest{}).(T)
	case *iop.WorkspaceArtifactRequest:
		return any(&iop.WorkspaceArtifactRequest{}).(T)
	case *iop.WorkspaceCancelRequest:
		return any(&iop.WorkspaceCancelRequest{}).(T)
	case *iop.WorkspaceCleanupRequest:
		return any(&iop.WorkspaceCleanupRequest{}).(T)
	case *iop.WorkspaceOpenResponse:
		return any(&iop.WorkspaceOpenResponse{}).(T)
	case *iop.WorkspaceToolResponse:
		return any(&iop.WorkspaceToolResponse{}).(T)
	case *iop.WorkspaceArtifactResponse:
		return any(&iop.WorkspaceArtifactResponse{}).(T)
	case *iop.WorkspaceCancelResponse:
		return any(&iop.WorkspaceCancelResponse{}).(T)
	case *iop.WorkspaceCleanupResponse:
		return any(&iop.WorkspaceCleanupResponse{}).(T)
	default:
		panic("unsupported workspace test message")
	}
}

func TestWorkspaceArtifactWire(t *testing.T) {
	t.Run("round trip and bounds", func(t *testing.T) {
		svc, node, binding := newWorkspaceWireFixture(t)
		var calls atomic.Int32
		seen := make(chan *iop.WorkspaceArtifactRequest, 2)
		toki.AddRequestListenerTyped[*iop.WorkspaceArtifactRequest, *iop.WorkspaceArtifactResponse](&node.Communicator, func(req *iop.WorkspaceArtifactRequest) (*iop.WorkspaceArtifactResponse, error) {
			calls.Add(1)
			seen <- proto.Clone(req).(*iop.WorkspaceArtifactRequest)
			response := &iop.WorkspaceArtifactResponse{RequestId: req.GetRequestId(), Kind: req.GetKind(), Operation: req.GetOperation(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}
			if req.GetOperation() == iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ {
				response.Content = []byte("bounded plan")
			}
			return response, nil
		})

		write := &iop.WorkspaceArtifactRequest{RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN, Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_WRITE, Content: []byte("bounded plan")}
		if response, err := svc.workspaceArtifact(context.Background(), binding, write, binding.Limits.MaxOutputBytes); err != nil || response.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS {
			t.Fatalf("write = %+v, %v", response, err)
		}
		write.Content[0] = 'X'
		if got := <-seen; string(got.GetContent()) != "bounded plan" {
			t.Fatalf("wire content changed after caller mutation: %q", got.GetContent())
		}
		read := &iop.WorkspaceArtifactRequest{RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN, Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ}
		if response, err := svc.workspaceArtifact(context.Background(), binding, read, binding.Limits.MaxOutputBytes); err != nil || string(response.GetContent()) != "bounded plan" {
			t.Fatalf("read = %+v, %v", response, err)
		}
		<-seen

		oversized := &iop.WorkspaceArtifactRequest{RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_REVIEW, Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_WRITE, Content: []byte(strings.Repeat("x", binding.Limits.MaxOutputBytes+1))}
		if _, err := svc.workspaceArtifact(context.Background(), binding, oversized, binding.Limits.MaxOutputBytes); !errors.Is(err, errWorkspaceWireArtifact) {
			t.Fatalf("oversized error = %v", err)
		}
		if got := calls.Load(); got != 2 {
			t.Fatalf("oversized write reached Node: calls=%d", got)
		}
	})

	for name, response := range map[string]*iop.WorkspaceArtifactResponse{
		"identity mismatch": {
			RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_REVIEW,
			Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ,
			Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
		},
		"oversized read": {
			RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN,
			Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ,
			Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Content: []byte(strings.Repeat("x", 65)),
		},
		"raw error": {
			RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN,
			Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ,
			Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
			Error: "raw node path sentinel",
		},
	} {
		t.Run(name, func(t *testing.T) {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceArtifactRequest, *iop.WorkspaceArtifactResponse](&node.Communicator, func(*iop.WorkspaceArtifactRequest) (*iop.WorkspaceArtifactResponse, error) {
				return proto.Clone(response).(*iop.WorkspaceArtifactResponse), nil
			})
			request := &iop.WorkspaceArtifactRequest{RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN, Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ}
			_, err := svc.workspaceArtifact(context.Background(), binding, request, binding.Limits.MaxOutputBytes)
			if !errors.Is(err, errWorkspaceWireResponse) || strings.Contains(err.Error(), "sentinel") {
				t.Fatalf("malformed response error = %v", err)
			}
		})
	}

	t.Run("stale generation", func(t *testing.T) {
		oldEdge, oldNode := workspaceWirePipe(t)
		newEdge, newNode := workspaceWirePipe(t)
		registry := edgenode.NewRegistry()
		oldEntry := &edgenode.NodeEntry{NodeID: "node-1", Client: oldEdge}
		registry.Register(oldEntry)
		binding := workspaceWireBinding("workspace-1", oldEntry.NodeID, oldEntry.ConnectionGeneration, 1000)
		registry.Register(&edgenode.NodeEntry{NodeID: "node-1", Client: newEdge})
		var reached atomic.Bool
		toki.AddRequestListenerTyped[*iop.WorkspaceArtifactRequest, *iop.WorkspaceArtifactResponse](&newNode.Communicator, func(req *iop.WorkspaceArtifactRequest) (*iop.WorkspaceArtifactResponse, error) {
			reached.Store(true)
			return &iop.WorkspaceArtifactResponse{RequestId: req.GetRequestId(), Kind: req.GetKind(), Operation: req.GetOperation(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}, nil
		})
		_ = oldNode
		svc := New(registry, edgeevents.NewBus())
		request := &iop.WorkspaceArtifactRequest{RequestId: "request-1", Kind: iop.WorkspaceArtifactKind_WORKSPACE_ARTIFACT_KIND_PLAN, Operation: iop.WorkspaceArtifactOperation_WORKSPACE_ARTIFACT_OPERATION_READ}
		if _, err := svc.workspaceArtifact(context.Background(), binding, request, binding.Limits.MaxOutputBytes); !errors.Is(err, errWorkspaceWireStale) {
			t.Fatalf("stale error = %v", err)
		}
		if reached.Load() {
			t.Fatal("stale artifact binding reselected reconnect client")
		}
	})
}

// newWorkspaceWireFixture wires an Edge Service to a single admitted Node over a
// net.Pipe and returns the Node communicator so a test can install responders.
func newWorkspaceWireFixture(t *testing.T) (*Service, *toki.TcpClient, *SingleRequestWorkspaceBinding) {
	t.Helper()
	edgeClient, nodeClient := workspaceWirePipe(t)
	registry := edgenode.NewRegistry()
	entry := &edgenode.NodeEntry{NodeID: "node-1", Client: edgeClient}
	registry.Register(entry)
	svc := New(registry, edgeevents.NewBus())
	binding := workspaceWireBinding("workspace-1", entry.NodeID, entry.ConnectionGeneration, 2000)
	return svc, nodeClient, binding
}

func workspaceWireBinding(ref, nodeID string, generation uint64, timeoutMS int) *SingleRequestWorkspaceBinding {
	return &SingleRequestWorkspaceBinding{
		Ref: ref, NodeID: nodeID, ConnectionGeneration: generation,
		OperationIDs: []string{"command", "read"}, CommandIDs: []string{"test"},
		Limits: SingleRequestWorkspaceLimits{MaxReadBytes: 64, MaxOutputBytes: 64, MaxCommandTimeoutMS: timeoutMS},
	}
}

// serveWorkspaceConcurrent installs a Node-side workspace responder that runs off
// the communicator's single receive coordinator, mirroring the Session's
// concurrent dispatch. Without this, a blocked tool responder would stall the
// coordinator and no queued cancel could be observed while the tool is in flight.
func serveWorkspaceConcurrent[Req proto.Message, Res proto.Message](c *toki.Communicator, seq *atomic.Int32, respond func(Req) Res) {
	c.AddRequestListener(toki.TypeNameOf(newWorkspaceMessage[Req]()), func(m proto.Message, requestNonce int32) {
		req, ok := m.(Req)
		if !ok {
			return
		}
		go func() {
			res := respond(req)
			data, err := proto.Marshal(res)
			if err != nil {
				return
			}
			_ = c.QueuePacket(&packets.PacketBase{TypeName: toki.TypeNameOf(res), Nonce: seq.Add(1), ResponseNonce: requestNonce, Data: data})
		}()
	})
}

// TestWorkspaceWireCancelReachesBlockedTool proves Required R1: a caller context
// cancelled while the Node tool handler is genuinely in flight delivers exactly
// one typed cancel, carrying the immutable identities, to the admitted Node
// before the tool handler is released — without the cancel waiting behind the
// registry dispatch-owner mutex held by the in-flight tool request.
func TestWorkspaceWireCancelReachesBlockedTool(t *testing.T) {
	svc, nodeClient, binding := newWorkspaceWireFixture(t)
	var seq atomic.Int32
	var cancelCount atomic.Int32
	toolEntered := make(chan struct{})
	cancelReached := make(chan *iop.WorkspaceCancelRequest, 1)
	release := make(chan struct{})
	serveWorkspaceConcurrent(&nodeClient.Communicator, &seq, func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
		close(toolEntered)
		<-release
		return &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}
	})
	serveWorkspaceConcurrent(&nodeClient.Communicator, &seq, func(req *iop.WorkspaceCancelRequest) *iop.WorkspaceCancelResponse {
		cancelCount.Add(1)
		cancelReached <- req
		return &iop.WorkspaceCancelResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_CANCELLED, Error: "workspace command cancelled"}
	})

	ctx, cancel := context.WithCancel(context.Background())
	toolErr := make(chan error, 1)
	go func() {
		_, err := svc.workspaceTool(ctx, binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
		toolErr <- err
	}()

	select {
	case <-toolEntered:
	case <-time.After(2 * time.Second):
		close(release)
		t.Fatal("node tool handler was never entered")
	}
	cancel()

	select {
	case got := <-cancelReached:
		if got.GetRequestId() != "request-1" || got.GetStageId() != "work" || got.GetToolCallId() != "tool-1" {
			close(release)
			t.Fatalf("cancel identity mismatch: %+v", got)
		}
	case <-time.After(time.Second):
		close(release)
		t.Fatal("cancel did not reach the blocked tool before release")
	}
	close(release)

	if err := <-toolErr; err == nil {
		t.Fatal("cancelled workspace tool must return an error")
	}
	// The cancel handler is exercised at most once; give a late duplicate time to
	// surface before asserting exactly-once delivery.
	time.Sleep(50 * time.Millisecond)
	if got := cancelCount.Load(); got != 1 {
		t.Fatalf("cancel sent %d times, want exactly 1", got)
	}
}

// TestWorkspaceWireRejectsBindingMismatchBeforeSend proves Required R2: an open
// request whose workspace reference is not the admitted reference fails closed
// before any transport dispatch and never reaches the Node.
func TestWorkspaceWireRejectsBindingMismatchBeforeSend(t *testing.T) {
	edgeClient, nodeClient := workspaceWirePipe(t)
	registry := edgenode.NewRegistry()
	entry := &edgenode.NodeEntry{NodeID: "node-1", Client: edgeClient}
	registry.Register(entry)
	svc := New(registry, edgeevents.NewBus())
	binding := workspaceWireBinding("approved", entry.NodeID, entry.ConnectionGeneration, 1000)
	var reached atomic.Bool
	toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&nodeClient.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
		reached.Store(true)
		return &iop.WorkspaceOpenResponse{RequestId: req.GetRequestId(), WorkspaceRef: req.GetWorkspaceRef(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}, nil
	})
	if _, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: "not-approved"}); !errors.Is(err, errWorkspaceWireReference) {
		t.Fatalf("mismatch error = %v, want %v", err, errWorkspaceWireReference)
	}
	time.Sleep(50 * time.Millisecond)
	if reached.Load() {
		t.Fatal("mismatched workspace reference must not reach the node")
	}
}

// TestWorkspaceWireRejectsInvalidResponse proves Required R3: every response
// family rejects a disallowed terminal status or a mismatched identity echo with
// a stable internal error, and never leaks the raw Node Error string.
func TestWorkspaceWireRejectsInvalidResponse(t *testing.T) {
	const rawSentinel = "RAW-NODE-ERROR-DO-NOT-LEAK-4711"
	cases := []struct {
		name string
		run  func(t *testing.T) error
	}{
		{"open status failure", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&node.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
				return &iop.WorkspaceOpenResponse{RequestId: req.GetRequestId(), WorkspaceRef: req.GetWorkspaceRef(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref})
			return err
		}},
		{"open identity mismatch", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&node.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
				return &iop.WorkspaceOpenResponse{RequestId: "other-request", WorkspaceRef: req.GetWorkspaceRef(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref})
			return err
		}},
		{"tool status failure", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&node.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
				return &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_TIMEOUT, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_TIMEOUT, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			return err
		}},
		{"tool identity mismatch", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&node.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
				return &iop.WorkspaceToolResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: "other-tool", Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			return err
		}},
		{"cancel wrong terminal", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCancelRequest, *iop.WorkspaceCancelResponse](&node.Communicator, func(req *iop.WorkspaceCancelRequest) (*iop.WorkspaceCancelResponse, error) {
				return &iop.WorkspaceCancelResponse{RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceCancel(context.Background(), binding, &iop.WorkspaceCancelRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			return err
		}},
		{"cleanup status failure", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCleanupRequest, *iop.WorkspaceCleanupResponse](&node.Communicator, func(req *iop.WorkspaceCleanupRequest) (*iop.WorkspaceCleanupResponse, error) {
				return &iop.WorkspaceCleanupResponse{RequestId: req.GetRequestId(), Status: iop.WorkspaceStatus_WORKSPACE_STATUS_UNSUPPORTED, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_NOT_READY, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceCleanup(context.Background(), binding, &iop.WorkspaceCleanupRequest{RequestId: "request-1"})
			return err
		}},
		{"cleanup identity mismatch", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCleanupRequest, *iop.WorkspaceCleanupResponse](&node.Communicator, func(req *iop.WorkspaceCleanupRequest) (*iop.WorkspaceCleanupResponse, error) {
				return &iop.WorkspaceCleanupResponse{RequestId: "other-request", Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Error: rawSentinel}, nil
			})
			_, err := svc.workspaceCleanup(context.Background(), binding, &iop.WorkspaceCleanupRequest{RequestId: "request-1"})
			return err
		}},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := tc.run(t)
			if err == nil {
				t.Fatal("invalid response must fail")
			}
			if !errors.Is(err, errWorkspaceWireResponse) {
				t.Fatalf("error = %v, want errWorkspaceWireResponse", err)
			}
			if strings.Contains(err.Error(), rawSentinel) {
				t.Fatalf("raw node error leaked in %q", err.Error())
			}
		})
	}
}

// TestWorkspaceWireBoundsRequestTimeout proves the request waiter is bounded: a
// Node that never responds fails the call well within the admitted command
// timeout rather than blocking indefinitely.
func TestWorkspaceWireBoundsRequestTimeout(t *testing.T) {
	edgeClient, nodeClient := workspaceWirePipe(t)
	registry := edgenode.NewRegistry()
	entry := &edgenode.NodeEntry{NodeID: "node-1", Client: edgeClient}
	registry.Register(entry)
	svc := New(registry, edgeevents.NewBus())
	binding := workspaceWireBinding("workspace-1", entry.NodeID, entry.ConnectionGeneration, 50)
	_ = nodeClient // no responder registered: the request must time out at the bound
	start := time.Now()
	if _, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref}); err == nil {
		t.Fatal("unanswered workspace open must fail")
	}
	if elapsed := time.Since(start); elapsed > time.Second {
		t.Fatalf("request wait exceeded bound: %v", elapsed)
	}
}

// TestWorkspaceWireRejectsContradictoryTerminalOutcome proves Required R3:
// an allowed terminal status paired with a failure error code or non-empty raw error
// returns nil response, returns errWorkspaceWireResponse, and does not leak raw error text.
func TestWorkspaceWireRejectsContradictoryTerminalOutcome(t *testing.T) {
	const rawSentinel = "RAW-CONTRADICTORY-ERROR-4711"
	cases := []struct {
		name string
		run  func(t *testing.T) error
	}{
		{"open success with error code", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&node.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
				return &iop.WorkspaceOpenResponse{
					RequestId:    req.GetRequestId(),
					WorkspaceRef: req.GetWorkspaceRef(),
					Status:       iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode:    iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
				}, nil
			})
			resp, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"open success with raw error text", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceOpenRequest, *iop.WorkspaceOpenResponse](&node.Communicator, func(req *iop.WorkspaceOpenRequest) (*iop.WorkspaceOpenResponse, error) {
				return &iop.WorkspaceOpenResponse{
					RequestId:    req.GetRequestId(),
					WorkspaceRef: req.GetWorkspaceRef(),
					Status:       iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode:    iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_UNSPECIFIED,
					Error:        rawSentinel,
				}, nil
			})
			resp, err := svc.workspaceOpen(context.Background(), binding, &iop.WorkspaceOpenRequest{RequestId: "request-1", WorkspaceRef: binding.Ref})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"tool success with error code", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&node.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
				return &iop.WorkspaceToolResponse{
					RequestId:  req.GetRequestId(),
					StageId:    req.GetStageId(),
					ToolCallId: req.GetToolCallId(),
					Status:     iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode:  iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
				}, nil
			})
			resp, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"tool success with raw error text", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&node.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
				return &iop.WorkspaceToolResponse{
					RequestId:  req.GetRequestId(),
					StageId:    req.GetStageId(),
					ToolCallId: req.GetToolCallId(),
					Status:     iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode:  iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_UNSPECIFIED,
					Error:      rawSentinel,
				}, nil
			})
			resp, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"cancel terminal with error code", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCancelRequest, *iop.WorkspaceCancelResponse](&node.Communicator, func(req *iop.WorkspaceCancelRequest) (*iop.WorkspaceCancelResponse, error) {
				return &iop.WorkspaceCancelResponse{
					RequestId:  req.GetRequestId(),
					StageId:    req.GetStageId(),
					ToolCallId: req.GetToolCallId(),
					Status:     iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED,
					ErrorCode:  iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
				}, nil
			})
			resp, err := svc.workspaceCancel(context.Background(), binding, &iop.WorkspaceCancelRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"cancel terminal with raw error text", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCancelRequest, *iop.WorkspaceCancelResponse](&node.Communicator, func(req *iop.WorkspaceCancelRequest) (*iop.WorkspaceCancelResponse, error) {
				return &iop.WorkspaceCancelResponse{
					RequestId:  req.GetRequestId(),
					StageId:    req.GetStageId(),
					ToolCallId: req.GetToolCallId(),
					Status:     iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED,
					ErrorCode:  iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_UNSPECIFIED,
					Error:      rawSentinel,
				}, nil
			})
			resp, err := svc.workspaceCancel(context.Background(), binding, &iop.WorkspaceCancelRequest{RequestId: "request-1", StageId: "work", ToolCallId: "tool-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"cleanup success with error code", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCleanupRequest, *iop.WorkspaceCleanupResponse](&node.Communicator, func(req *iop.WorkspaceCleanupRequest) (*iop.WorkspaceCleanupResponse, error) {
				return &iop.WorkspaceCleanupResponse{
					RequestId: req.GetRequestId(),
					Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_INTERNAL,
				}, nil
			})
			resp, err := svc.workspaceCleanup(context.Background(), binding, &iop.WorkspaceCleanupRequest{RequestId: "request-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
		{"cleanup success with raw error text", func(t *testing.T) error {
			svc, node, binding := newWorkspaceWireFixture(t)
			toki.AddRequestListenerTyped[*iop.WorkspaceCleanupRequest, *iop.WorkspaceCleanupResponse](&node.Communicator, func(req *iop.WorkspaceCleanupRequest) (*iop.WorkspaceCleanupResponse, error) {
				return &iop.WorkspaceCleanupResponse{
					RequestId: req.GetRequestId(),
					Status:    iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
					ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_UNSPECIFIED,
					Error:     rawSentinel,
				}, nil
			})
			resp, err := svc.workspaceCleanup(context.Background(), binding, &iop.WorkspaceCleanupRequest{RequestId: "request-1"})
			if resp != nil {
				t.Fatalf("expected nil response, got %+v", resp)
			}
			return err
		}},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := tc.run(t)
			if err == nil {
				t.Fatal("contradictory terminal outcome must fail")
			}
			if !errors.Is(err, errWorkspaceWireResponse) {
				t.Fatalf("error = %v, want errWorkspaceWireResponse", err)
			}
			if strings.Contains(err.Error(), rawSentinel) {
				t.Fatalf("raw node error leaked in %q", err.Error())
			}
		})
	}
}

// TestWorkspaceWireAcceptsCanonicalNonSuccessResponses proves that Edge accepts
// canonical non-success tool and cancel responses and retains bounded fields.
func TestWorkspaceWireAcceptsCanonicalNonSuccessResponses(t *testing.T) {
	svc, node, binding := newWorkspaceWireFixture(t)
	toki.AddRequestListenerTyped[*iop.WorkspaceToolRequest, *iop.WorkspaceToolResponse](&node.Communicator, func(req *iop.WorkspaceToolRequest) (*iop.WorkspaceToolResponse, error) {
		switch req.GetToolCallId() {
		case "nonzero":
			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: "workspace operation failed", Stdout: []byte("out"), Stderr: []byte("err"), ExitCode: 7, DurationMs: 50,
			}, nil
		case "timeout":
			return &iop.WorkspaceToolResponse{
				RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
				Status: iop.WorkspaceStatus_WORKSPACE_STATUS_TIMEOUT, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_TIMEOUT,
				Error: "workspace command timed out", ExitCode: -1, DurationMs: 2000,
			}, nil
		case "cancelled":
			return &iop.WorkspaceToolResponse{
				RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
				Status: iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_CANCELLED,
				Error: "workspace command cancelled", ExitCode: -1,
			}, nil
		default:
			return nil, errors.New("unknown tool call")
		}
	})

	nonzero, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "nonzero"})
	if err != nil || nonzero.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_ERROR || nonzero.GetExitCode() != 7 || string(nonzero.GetStdout()) != "out" || string(nonzero.GetStderr()) != "err" {
		t.Fatalf("nonzero response = %+v, %v", nonzero, err)
	}

	timeout, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "timeout"})
	if err != nil || timeout.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_TIMEOUT || timeout.GetErrorCode() != iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_TIMEOUT || timeout.GetExitCode() != -1 {
		t.Fatalf("timeout response = %+v, %v", timeout, err)
	}

	cancelled, err := svc.workspaceTool(context.Background(), binding, &iop.WorkspaceToolRequest{RequestId: "request-1", StageId: "work", ToolCallId: "cancelled"})
	if err != nil || cancelled.GetStatus() != iop.WorkspaceStatus_WORKSPACE_STATUS_CANCELLED || cancelled.GetErrorCode() != iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_CANCELLED {
		t.Fatalf("cancelled response = %+v, %v", cancelled, err)
	}
}