Remove unreachable legacy CLI entry points

cmd/sip-go-agent/dispatcher_command_test.go

@@ -1,46 +0,0 @@
-package main
-
-import (
-	"errors"
-	"io"
-	"os"
-	"path/filepath"
-	"strings"
-	"testing"
-)
-
-func TestDispatcherCommandDoesNotExposeLegacyTenantQueueFlags(t *testing.T) {
-	cmd := newDispatcherCommand()
-	for _, name := range []string{"tenant-key", "consume"} {
-		if cmd.Flags().Lookup(name) != nil {
-			t.Errorf("legacy flag --%s is still exposed", name)
-		}
-	}
-}
-
-func TestDispatcherRejectsNonMockExecutionBeforeOpeningResources(t *testing.T) {
-	for _, mode := range []string{"mixed", "real"} {
-		t.Run(mode, func(t *testing.T) {
-			dbPath := filepath.Join(t.TempDir(), "must-not-open.sqlite")
-			t.Setenv("SIP_GO_AGENT_MODE", mode)
-			t.Setenv("DISPATCHER_DB", dbPath)
-			t.Setenv("RABBITMQ_URL", "amqp://127.0.0.1:1/")
-			t.Setenv("DISPATCHER_GRPC_LISTEN", "")
-			t.Setenv("DISPATCHER_AGENT_ENDPOINTS_FILE", "")
-			t.Setenv("GO_SIP_OSS_ACCESS_KEY_ID", "local-test-placeholder")
-			t.Setenv("GO_SIP_OSS_ACCESS_KEY_SECRET", "local-test-placeholder")
-			cmd := newDispatcherCommand()
-			cmd.SetArgs([]string{"--config", filepath.Join("..", "..", "deploys", "config", "dispatcher.json.example")})
-			cmd.SilenceUsage, cmd.SilenceErrors = true, true
-			cmd.SetOut(io.Discard)
-			cmd.SetErr(io.Discard)
-			err := cmd.Execute()
-			if err == nil || !strings.Contains(err.Error(), "isolated Mock only") {
-				t.Fatalf("%s dispatcher reached resources instead of refusing unapproved execution: %v", mode, err)
-			}
-			if _, err := os.Stat(dbPath); !errors.Is(err, os.ErrNotExist) {
-				t.Fatalf("%s dispatcher opened a business DB before authorization: %v", mode, err)
-			}
-		})
-	}
-}

cmd/sip-go-agent/dispatcher_config_test.go

@@ -6,9 +6,6 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
-
-	"git.ipao.vip/rogee/go-sip/internal/config"
-	"git.ipao.vip/rogee/go-sip/internal/store"
 )
 
 func TestDispatcherHasNoBusinessHTTPFlags(t *testing.T) {
@@ -33,39 +30,3 @@ func TestDispatcherRequiresDeploymentIdentityBeforeOpeningDatabase(t *testing.T)
 		t.Fatalf("database was touched before deployment validation: %v", err)
 	}
 }
-
-func TestDispatcherStoreRejectsChangedIdentityBeforeRecovery(t *testing.T) {
-	path := filepath.Join(t.TempDir(), "identity.db")
-	cfg := config.Config{DBPath: path, DispatcherID: "c046b893-8628-4589-ae50-619d049248a6"}
-	st, err := openDispatcherStore(cfg)
-	if err != nil {
-		t.Fatal(err)
-	}
-	// A claimed notice must not be reset by a process presenting another ID.
-	_, err = st.DB().Exec(`INSERT INTO outbox (event_id, tenant_key, exchange, routing_key, body, status, created_at) VALUES ('notice','key','exchange','route','{}','dispatching','2026-09-21T00:00:00Z')`)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err := st.Close(); err != nil {
-		t.Fatal(err)
-	}
-	cfg.DispatcherID = "a50b1569-aa17-4503-bfc1-cf55c10a1c24"
-	if other, err := openDispatcherStore(cfg); !errors.Is(err, store.ErrDispatcherIdentityMismatch) {
-		if other != nil {
-			other.Close()
-		}
-		t.Fatalf("wrong identity accepted: %v", err)
-	}
-	check, err := store.Open(path)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer check.Close()
-	var status string
-	if err := check.DB().QueryRow(`SELECT status FROM outbox WHERE event_id='notice'`).Scan(&status); err != nil {
-		t.Fatal(err)
-	}
-	if status != "dispatching" {
-		t.Fatalf("foreign identity changed recovery state: %s", status)
-	}
-}

cmd/sip-go-agent/main.go

@@ -1,41 +1,10 @@
 package main
 
 import (
-	"context"
-	"encoding/json"
-	"errors"
-	"fmt"
 	"log/slog"
-	"net"
-	"net/http"
 	"os"
-	"os/signal"
-	"path/filepath"
-	"strings"
-	"syscall"
-	"time"
 
-	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
-	"git.ipao.vip/rogee/go-sip/internal/agent"
-	"git.ipao.vip/rogee/go-sip/internal/ai"
-	"git.ipao.vip/rogee/go-sip/internal/callflow"
-	"git.ipao.vip/rogee/go-sip/internal/calllog"
-	"git.ipao.vip/rogee/go-sip/internal/callruntime"
-	"git.ipao.vip/rogee/go-sip/internal/callwindow"
-	"git.ipao.vip/rogee/go-sip/internal/config"
-	"git.ipao.vip/rogee/go-sip/internal/configread"
-	"git.ipao.vip/rogee/go-sip/internal/contract"
-	"git.ipao.vip/rogee/go-sip/internal/dispatcher"
-	"git.ipao.vip/rogee/go-sip/internal/health"
-	"git.ipao.vip/rogee/go-sip/internal/mq"
-	ossclient "git.ipao.vip/rogee/go-sip/internal/oss"
-	"git.ipao.vip/rogee/go-sip/internal/rpc"
-	"git.ipao.vip/rogee/go-sip/internal/store"
 	"github.com/spf13/cobra"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/status"
 )
 
 func main() {
@@ -55,673 +24,3 @@ func newRootCommand() *cobra.Command {
 	root.AddCommand(newCurrentAgentCommand(), newCurrentDispatcherCommand())
 	return root
 }
-
-func newAgentCommand() *cobra.Command {
-	cfg, configErr := config.FromEnv()
-	var realCall bool
-	cmd := &cobra.Command{
-		Use:   "agent",
-		Short: "run the file-backed Agent process",
-		RunE: func(_ *cobra.Command, _ []string) error {
-			if configErr != nil {
-				return configErr
-			}
-			if realCall {
-				return runCallOnce(cfg)
-			}
-			if err := cfg.Validate("agent"); err != nil {
-				return err
-			}
-			spool, err := agent.NewSpool(cfg.SpoolRoot, nil)
-			if err != nil {
-				return err
-			}
-			report, err := spool.MarkUnknownOnBoot()
-			if err != nil {
-				return err
-			}
-			if cfg.GRPCListen == "" {
-				return writeResult(map[string]any{
-					"role": "agent", "mode": cfg.Mode, "agent_id": cfg.AgentID,
-					"version": cfg.Version, "spool": spool.Root(),
-					"unknown_executions": report.Unknown, "quarantined": report.Quarantined,
-				})
-			}
-			return serveAgentRPC(cfg, spool, report)
-		},
-	}
-	cmd.Flags().StringVar(&cfg.Mode, "mode", cfg.Mode, "mock, mixed, or real")
-	cmd.Flags().StringVar(&cfg.SpoolRoot, "spool", cfg.SpoolRoot, "Agent file spool root")
-	cmd.Flags().StringVar(&cfg.AgentID, "agent-id", cfg.AgentID, "stable Agent identifier")
-	cmd.Flags().StringVar(&cfg.CellID, "cell-id", cfg.CellID, "stable Cell identifier")
-	cmd.Flags().StringVar(&cfg.Version, "version", cfg.Version, "Agent software version")
-	cmd.Flags().StringVar(&cfg.StaticArtifactPath, "static-artifact", cfg.StaticArtifactPath, "management-approved static Cell artifact path")
-	cmd.Flags().BoolVar(&realCall, "call-once", false, "run one explicit call through the configured transport and AI adapters")
-	cmd.Flags().StringVar(&cfg.CallTarget, "call-target", cfg.CallTarget, "raw target number allowed by the static artifact")
-	cmd.Flags().StringVar(&cfg.CallTrunkID, "call-trunk-id", cfg.CallTrunkID, "enabled trunk ID from the static artifact")
-	cmd.Flags().StringVar(&cfg.CallCallerID, "call-caller-id", cfg.CallCallerID, "deployment-approved caller ID")
-	cmd.Flags().StringVar(&cfg.CallAISnapshotPath, "call-ai-snapshot", cfg.CallAISnapshotPath, "immutable AI snapshot path")
-	cmd.Flags().IntVar(&cfg.CallMediaPort, "call-media-port", cfg.CallMediaPort, "ExternalMedia UDP port")
-	cmd.Flags().StringVar(&cfg.CallRecordingDirectory, "call-recording-dir", cfg.CallRecordingDirectory, "local recording directory")
-	cmd.AddCommand(newUploadRetryCommand())
-	return cmd
-}
-
-func mediaForCall(artifact contract.StaticCellArtifact, trunkID string) (contract.StaticMedia, error) {
-	if artifact.Media == nil {
-		return contract.StaticMedia{}, errors.New("static artifact has no default media profile")
-	}
-	for _, trunk := range artifact.Trunks {
-		if trunk.TrunkID != trunkID || !trunk.Enabled {
-			continue
-		}
-		media := *artifact.Media
-		if trunk.MediaProfileID == "" {
-			return media, nil
-		}
-		profile, ok := artifact.MediaProfiles[trunk.MediaProfileID]
-		if !ok {
-			return contract.StaticMedia{}, fmt.Errorf("call trunk %q references unknown media profile %q", trunkID, trunk.MediaProfileID)
-		}
-		media.Format = profile.Format
-		media.SampleRateHz = profile.SampleRateHz
-		media.Channels = profile.Channels
-		media.PayloadType = profile.PayloadType
-		return media, nil
-	}
-	return contract.StaticMedia{}, fmt.Errorf("call trunk %q is not enabled in the static artifact", trunkID)
-}
-
-func buildCallEndpoint(artifact contract.StaticCellArtifact, trunkID, target string) (string, error) {
-	if target == "" {
-		return "", errors.New("call target is required")
-	}
-	allowedTarget := false
-	for _, allowed := range artifact.AllowedTargets {
-		if allowed == target {
-			allowedTarget = true
-			break
-		}
-	}
-	if !allowedTarget {
-		return "", fmt.Errorf("call target is not allowed by the static artifact")
-	}
-	for _, trunk := range artifact.Trunks {
-		if trunk.TrunkID != trunkID || !trunk.Enabled {
-			continue
-		}
-		if trunk.SIPEndpointRef == "" {
-			return "", errors.New("selected trunk has no SIP endpoint reference")
-		}
-		return fmt.Sprintf("PJSIP/%s%s@%s", trunk.DialPrefix, target, trunk.SIPEndpointRef), nil
-	}
-	return "", fmt.Errorf("call trunk %q is not enabled in the static artifact", trunkID)
-}
-
-func validateCallAISnapshot(raw []byte) (ai.Snapshot, error) {
-	snapshot, err := ai.Validate(raw)
-	if err != nil {
-		return ai.Snapshot{}, fmt.Errorf("validate AI snapshot: %w", err)
-	}
-	return snapshot, nil
-}
-
-func runCallOnce(cfg config.Config) error {
-	if cfg.StaticArtifactPath == "" || cfg.CallAISnapshotPath == "" {
-		return errors.New("--call-once requires --static-artifact and --call-ai-snapshot")
-	}
-	artifactRaw, err := os.ReadFile(cfg.StaticArtifactPath)
-	if err != nil {
-		return fmt.Errorf("read static artifact: %w", err)
-	}
-	artifact, err := contract.ValidateStaticArtifact(artifactRaw, contract.StaticArtifactExpectation{CellID: cfg.CellID, Mode: cfg.Mode})
-	if err != nil {
-		return fmt.Errorf("validate static artifact: %w", err)
-	}
-	if cfg.CallTrunkID == "" {
-		return errors.New("--call-trunk-id is required")
-	}
-	trunkBound := false
-	for _, trunk := range artifact.Trunks {
-		if trunk.TrunkID == cfg.CallTrunkID && trunk.Enabled {
-			trunkBound = true
-			break
-		}
-	}
-	if !trunkBound {
-		return fmt.Errorf("call trunk %q is not enabled in the static artifact", cfg.CallTrunkID)
-	}
-	if cfg.Mode != "mock" {
-		if err := callwindow.Check(time.Now()); err != nil {
-			return err
-		}
-	}
-	snapshotRaw, err := os.ReadFile(cfg.CallAISnapshotPath)
-	if err != nil {
-		return fmt.Errorf("read immutable AI snapshot: %w", err)
-	}
-	snapshot, err := validateCallAISnapshot(snapshotRaw)
-	if err != nil {
-		return err
-	}
-	var snapshotPolicy struct {
-		Conversation struct {
-			Opening   string `json:"opening"`
-			MaxTurns  int    `json:"max_turns"`
-			MaxMillis int    `json:"max_duration_ms"`
-		} `json:"conversation"`
-	}
-	if err := json.Unmarshal(snapshotRaw, &snapshotPolicy); err != nil {
-		return fmt.Errorf("read conversation policy from AI snapshot: %w", err)
-	}
-	conversation := snapshotPolicy.Conversation
-
-	var pipeline ai.Pipeline
-	switch cfg.Mode {
-	case "mock", "mixed":
-		pipeline = ai.MockPipeline{MaxAudioBytes: 16 << 20}
-	case "real":
-		providerCfg, providerErr := ai.LoadProviderPipelineConfigFromEnv()
-		if providerErr != nil {
-			return providerErr
-		}
-		pipeline, err = ai.NewProviderPipeline(providerCfg)
-		if err != nil {
-			return err
-		}
-	default:
-		return fmt.Errorf("unsupported call mode %q", cfg.Mode)
-	}
-	callCtx, cancel := context.WithTimeout(context.Background(), 4*time.Minute)
-	defer cancel()
-	if cfg.Mode == "mock" {
-		input := make([]byte, 6400)
-		flowResult, flowErr := callflow.Execute(callCtx, callflow.NewMemorySession(input), pipeline, snapshot, "您好，这是测试流程。", 10*time.Millisecond)
-		if flowErr != nil {
-			return flowErr
-		}
-		return writeResult(map[string]any{
-			"role": "agent", "mode": cfg.Mode, "ai_mode": snapshot.Mode, "call": "completed",
-			"transcript_chars": len([]rune(flowResult.Turn.Transcript)),
-			"reply_chars":      len([]rune(flowResult.Turn.Reply)),
-			"rtp":              map[string]any{"received_packets": flowResult.RTP.ReceivedPackets, "received_bytes": flowResult.RTP.ReceivedBytes, "sent_packets": flowResult.RTP.SentPackets, "sent_bytes": flowResult.RTP.SentBytes},
-		})
-	}
-	if artifact.Media == nil || artifact.ARI == nil || artifact.Recording == nil {
-		return errors.New("static artifact is missing transport/media/recording sections")
-	}
-	if cfg.CallTarget == "" || cfg.ARIUsername == "" || cfg.ARIPassword == "" {
-		return errors.New("--call-once requires target and ARI credentials")
-	}
-	callEndpoint, err := buildCallEndpoint(artifact, cfg.CallTrunkID, cfg.CallTarget)
-	if err != nil {
-		return err
-	}
-	selectedMedia, err := mediaForCall(artifact, cfg.CallTrunkID)
-	if err != nil {
-		return err
-	}
-	if cfg.CallMediaPort == 0 {
-		cfg.CallMediaPort = selectedMedia.Port
-	}
-	recordingDir := cfg.CallRecordingDirectory
-	if recordingDir == "" {
-		recordingDir = artifact.Recording.Directory
-	}
-	maxCallDuration := time.Duration(conversation.MaxMillis) * time.Millisecond
-	result, err := callruntime.Run(callCtx, callruntime.Config{
-		ARIURL: cfg.ARIURL, ARIWebsocketURL: cfg.ARIWebsocketURL, ARIApplication: artifact.ARI.Application,
-		ARIUsername: cfg.ARIUsername, ARIPassword: cfg.ARIPassword, Endpoint: callEndpoint,
-		CallerID: cfg.CallCallerID, MediaBind: selectedMedia.BindAddress, MediaPort: cfg.CallMediaPort,
-		MediaFormat: selectedMedia.Format, MediaSampleRate: selectedMedia.SampleRateHz,
-		PayloadType: uint8(selectedMedia.PayloadType), RecordingDirectory: recordingDir,
-		MaxTurns: conversation.MaxTurns, MaxCallDuration: maxCallDuration, OpeningPrompt: conversation.Opening,
-		Snapshot: snapshot, Pipeline: pipeline,
-	})
-	if err != nil {
-		return err
-	}
-	turns := make([]map[string]any, 0, len(result.Turns))
-	includeText := os.Getenv("AGENT_CALL_INCLUDE_TEXT") == "1"
-	for index, turn := range result.Turns {
-		fact := map[string]any{
-			"turn":             index + 1,
-			"transcript_chars": len([]rune(turn.Transcript)),
-			"reply_chars":      len([]rune(turn.Reply)),
-			"invalid_call":     turn.InvalidCall,
-			"invalid_reason":   turn.InvalidReason,
-		}
-		if includeText {
-			fact["transcript"] = turn.Transcript
-			fact["reply"] = turn.Reply
-		}
-		turns = append(turns, fact)
-	}
-	inboundRecordings := make([]map[string]any, 0, len(result.InboundRecordings))
-	for _, recording := range result.InboundRecordings {
-		inboundRecordings = append(inboundRecordings, map[string]any{"segment": recording.Segment, "path": recording.Path, "bytes": recording.Bytes, "sha256": recording.SHA256})
-	}
-	outboundRecordings := make([]map[string]any, 0, len(result.OutboundRecordings))
-	for _, recording := range result.OutboundRecordings {
-		outboundRecordings = append(outboundRecordings, map[string]any{"segment": recording.Segment, "path": recording.Path, "bytes": recording.Bytes, "sha256": recording.SHA256})
-	}
-	uploads, err := uploadCallRecordings(callCtx, cfg, result)
-	if err != nil {
-		return err
-	}
-	output := map[string]any{
-		"role": "agent", "mode": cfg.Mode, "call": "completed", "channel_id": result.ChannelID,
-		"transcript_chars": len([]rune(result.Transcript)), "reply_chars": len([]rune(result.Reply)),
-		"invalid_call": result.InvalidCall, "invalid_reason": result.InvalidReason,
-		"turns":              turns,
-		"inbound_recording":  map[string]any{"path": result.InboundPath, "bytes": result.InboundBytes, "sha256": result.InboundSHA256},
-		"outbound_recording": map[string]any{"path": result.OutboundPath, "bytes": result.OutboundBytes, "sha256": result.OutboundSHA256},
-		"inbound_recordings": inboundRecordings, "outbound_recordings": outboundRecordings,
-		"rtp": map[string]any{"received_packets": result.RTP.ReceivedPackets, "received_bytes": result.RTP.ReceivedBytes, "sent_packets": result.RTP.SentPackets, "sent_bytes": result.RTP.SentBytes},
-	}
-	if uploads != nil {
-		output["oss_uploads"] = uploads
-	}
-	return writeResult(output)
-}
-
-func serveAgentRPC(cfg config.Config, spool *agent.Spool, report agent.RecoveryReport) error {
-	var staticArtifactRaw []byte
-	if cfg.StaticArtifactPath != "" {
-		var err error
-		staticArtifactRaw, err = os.ReadFile(cfg.StaticArtifactPath)
-		if err != nil {
-			return fmt.Errorf("read static Cell artifact: %w", err)
-		}
-	}
-	peerCertificateFingerprints, err := config.ParseCertificateFingerprints(cfg.MTLSPeerCertificateFingerprints)
-	if err != nil {
-		return fmt.Errorf("parse MTLS_PEER_CERT_FINGERPRINTS: %w", err)
-	}
-	tlsConfig, err := rpc.LoadServerTLSConfig(cfg.MTLSCAFile, cfg.MTLSCertFile, cfg.MTLSKeyFile)
-	if err != nil {
-		return err
-	}
-	var callLogger *calllog.Logger
-	if cfg.CallLogPhoneKey != "" {
-		path := cfg.CallLogPath
-		if path == "" {
-			path = filepath.Join(spool.Root(), "call-business.jsonl")
-		}
-		callLogger, err = calllog.New(path, []byte(cfg.CallLogPhoneKey), nil)
-		if err != nil {
-			return fmt.Errorf("configure call business log: %w", err)
-		}
-	}
-	listener, err := net.Listen("tcp", cfg.GRPCListen)
-	if err != nil {
-		return fmt.Errorf("listen Agent gRPC: %w", err)
-	}
-	defer listener.Close()
-
-	bootID := fmt.Sprintf("%s-%d", cfg.AgentID, time.Now().UnixNano())
-	grpcServer := grpc.NewServer(grpc.Creds(credentials.NewTLS(tlsConfig)))
-	handler := rpc.NewServer(rpc.ServerOptions{
-		Mode: cfg.Mode,
-		Status: &agentpb.AgentStatus{
-			AgentId:         cfg.AgentID,
-			CellId:          cfg.CellID,
-			BootId:          bootID,
-			SoftwareVersion: cfg.Version,
-			ProtocolVersion: "agent.v1",
-			AdmissionState:  agentpb.AdmissionState_ADMISSION_STATE_CLOSED,
-			StatusReason:    fmt.Sprintf("recovered_unknown=%d quarantined=%d spool=%s", len(report.Unknown), len(report.Quarantined), spool.Root()),
-			Capabilities:    []*agentpb.Capability{{Name: "mode", Value: cfg.Mode}, {Name: "grpc_transport", Value: "unary-mtls"}, {Name: "resource_sample", Value: "partial-unknown"}},
-			Resources:       health.Sampler{}.Sample(context.Background(), spool.Root()),
-		},
-		UploadPolicy:                &agentpb.UploadPolicy{Enabled: cfg.Mode == "mock", MaxAssetBytes: 16 << 20},
-		StaticArtifactRaw:           staticArtifactRaw,
-		StaticArtifactExpected:      contract.StaticArtifactExpectation{CellID: cfg.CellID, Mode: cfg.Mode},
-		RequirePeerCertificate:      true,
-		PeerCertificateFingerprints: peerCertificateFingerprints,
-		StatePath:                   filepath.Join(cfg.SpoolRoot, "rpc-session.json"),
-		CallLogger:                  callLogger,
-		MockAuthorizedOriginate:     mockAuthorizedOriginator(cfg.Mode),
-	})
-	agentpb.RegisterAgentControlServiceServer(grpcServer, handler)
-	serveCtx, cancel := signalContext()
-	defer cancel()
-	stopUploadRecovery, err := startUploadNotificationRecovery(serveCtx, cfg, spool, handler.ActiveSessionMeta)
-	if err != nil {
-		return err
-	}
-	defer stopUploadRecovery()
-	go func() {
-		<-serveCtx.Done()
-		grpcServer.GracefulStop()
-	}()
-	if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-		return err
-	}
-	return nil
-}
-
-type connectedAgentRuntime struct {
-	coordinator *dispatcher.AgentCoordinator
-	clients     []*rpc.Client
-	sessions    []dispatcher.AgentSession
-}
-
-// connectConfiguredAgents performs the Dispatcher startup binding for the
-// deployment-owned endpoint inventory. It is intentionally separate from SaaS
-// commands: tenant input never selects an endpoint or identity.
-func connectConfiguredAgents(ctx context.Context, cfg config.Config) (*connectedAgentRuntime, error) {
-	endpoints, err := config.LoadAgentEndpoints(cfg.AgentEndpointsFile)
-	if err != nil {
-		return nil, err
-	}
-	runtime := &connectedAgentRuntime{coordinator: dispatcher.NewAgentCoordinator(nil)}
-	fail := func(err error) (*connectedAgentRuntime, error) {
-		return nil, errors.Join(err, runtime.Close())
-	}
-	epoch := fmt.Sprintf("%s-%d", cfg.DispatcherID, time.Now().UnixNano())
-	for _, endpoint := range endpoints {
-		client, dialErr := rpc.DialFromFiles(endpoint.Address, cfg.MTLSCAFile, cfg.MTLSCertFile, cfg.MTLSKeyFile, endpoint.ServerName)
-		if dialErr != nil {
-			return fail(fmt.Errorf("connect Agent %q: %w", endpoint.AgentID, dialErr))
-		}
-		runtime.clients = append(runtime.clients, client)
-		if registerErr := runtime.coordinator.Register(endpoint.AgentID, client.Agent); registerErr != nil {
-			return fail(fmt.Errorf("register Agent %q: %w", endpoint.AgentID, registerErr))
-		}
-		status, probeErr := runtime.coordinator.Probe(ctx, endpoint.AgentID, endpoint.CellID)
-		if probeErr != nil {
-			return fail(fmt.Errorf("probe Agent %q: %w", endpoint.AgentID, probeErr))
-		}
-		// Zero lets Agent-side durable session state allocate the next generation
-		// after a Dispatcher restart; hard-coding 1 would self-fence recovery.
-		session, activateErr := runtime.coordinator.Activate(ctx, cfg.DispatcherID, endpoint.AgentID, endpoint.CellID, status.BootId, epoch, 0)
-		if activateErr != nil {
-			return fail(fmt.Errorf("activate Agent %q: %w", endpoint.AgentID, activateErr))
-		}
-		runtime.sessions = append(runtime.sessions, session)
-	}
-	return runtime, nil
-}
-
-func (r *connectedAgentRuntime) Close() error {
-	if r == nil {
-		return nil
-	}
-	var firstErr error
-	for index := len(r.clients) - 1; index >= 0; index-- {
-		if err := r.clients[index].Close(); err != nil && firstErr == nil {
-			firstErr = err
-		}
-	}
-	return firstErr
-}
-
-func openDispatcherStore(cfg config.Config) (*store.Store, error) {
-	st, err := store.Open(cfg.DBPath)
-	if err != nil {
-		return nil, err
-	}
-	if err := st.BindDispatcherID(cfg.DispatcherID); err != nil {
-		return nil, errors.Join(err, st.Close())
-	}
-	if err := st.RecoverOutbox(); err != nil {
-		return nil, errors.Join(err, st.Close())
-	}
-	return st, nil
-}
-
-func newDispatcherCommand() *cobra.Command {
-	cfg, configErr := config.FromEnv()
-	var configFile string
-	var once bool
-	cmd := &cobra.Command{
-		Use:   "dispatcher",
-		Short: "run the single-active Dispatcher process",
-		RunE: func(cmd *cobra.Command, _ []string) error {
-			if configErr != nil {
-				return configErr
-			}
-			if err := cfg.LoadDispatcherFile(configFile); err != nil {
-				return err
-			}
-			if err := cfg.Validate("dispatcher"); err != nil {
-				return err
-			}
-			if cfg.Mode != "mock" && !once {
-				return errors.New("Dispatcher V3 authorized execution and upload are isolated Mock only; mixed/real requires separate approval and implementation")
-			}
-			var publisher mq.Publisher
-			var broker *mq.V3Broker
-			if cfg.RabbitURL != "" {
-				var err error
-				broker, err = mq.OpenV3(cfg.RabbitURL, cfg.DispatcherID, 1)
-				if err != nil {
-					return err
-				}
-				defer broker.Close()
-				publisher = broker
-			}
-			st, err := openDispatcherStore(cfg)
-			if err != nil {
-				return err
-			}
-			defer st.Close()
-			signalCtx, cancelSignal := signalContext()
-			defer cancelSignal()
-			leaseCtx := signalCtx
-			if broker != nil {
-				var cancelIdentity context.CancelFunc
-				leaseCtx, cancelIdentity = dispatcherIdentityContext(signalCtx, broker.Done())
-				defer cancelIdentity()
-			}
-			lease, err := dispatcher.StartLease(leaseCtx, st, "dispatcher-active-"+cfg.DispatcherID, "dispatcher", cfg.DispatcherID, 30*time.Second)
-			if err != nil {
-				return err
-			}
-			defer lease.Stop()
-			var connectedAgents *connectedAgentRuntime
-			if cfg.AgentEndpointsFile != "" {
-				startupCtx, cancelStartup := context.WithTimeout(leaseCtx, 15*time.Second)
-				connectedAgents, err = connectConfiguredAgents(startupCtx, cfg)
-				cancelStartup()
-				if err != nil {
-					return err
-				}
-				defer func() {
-					if closeErr := connectedAgents.Close(); closeErr != nil {
-						slog.Error("close Agent connections", "error", closeErr)
-					}
-				}()
-			}
-			d, err := dispatcher.NewV3(cfg.DispatcherID, st, publisher, time.Now)
-			if err != nil {
-				return err
-			}
-			var dispatcherGRPC *grpc.Server
-			var dispatcherListener net.Listener
-			if cfg.DispatcherGRPCListen != "" && !once {
-				uploadClient, ossErr := ossclient.NewClient(ossclient.Config{
-					Endpoint: cfg.OSSEndpoint, Region: cfg.OSSRegion, Bucket: cfg.OSSBucket,
-					AccessKeyID: cfg.OSSAccessKeyID, AccessKeySecret: cfg.OSSAccessKeySecret,
-					KeyPrefix: cfg.OSSKeyPrefix, GrantTTL: cfg.OSSGrantTTL, MaxAssetBytes: cfg.OSSMaxAssetBytes,
-				})
-				if ossErr != nil {
-					return fmt.Errorf("configure Dispatcher OSS: %w", ossErr)
-				}
-				peerFingerprints, fingerprintErr := config.ParseCertificateFingerprints(cfg.MTLSPeerCertificateFingerprints)
-				if fingerprintErr != nil {
-					return fmt.Errorf("parse Dispatcher gRPC mTLS peer fingerprints: %w", fingerprintErr)
-				}
-				if len(peerFingerprints) == 0 {
-					return errors.New("MTLS_PEER_CERT_FINGERPRINTS is required when Dispatcher gRPC is enabled")
-				}
-				allowedAgentIDs := parseCSVSet(cfg.DispatcherGRPCAllowedAgentIDs)
-				authorizeAgentSession := func(meta *agentpb.RequestMeta) error {
-					if connectedAgents == nil || connectedAgents.coordinator == nil {
-						return fmt.Errorf("no activated Agent coordinator: %w", store.ErrCommandConflict)
-					}
-					return connectedAgents.coordinator.AuthorizeInboundMeta(meta)
-				}
-				uploadHandler, handlerErr := rpc.NewDispatcherUploadServerWithOptions(st, uploadClient, time.Now, rpc.DispatcherUploadOptions{
-					RequirePeer: true, PeerCertificateFingerprints: peerFingerprints, AllowedAgentIDs: allowedAgentIDs,
-					LocalV3Authorize: func(ctx context.Context, req *agentpb.RequestUploadRequest) error {
-						if err := authorizeAgentSession(req.Meta); err != nil {
-							return err
-						}
-						return d.AuthorizeLocalMockUpload(ctx, req)
-					},
-					LocalV3Complete: func(ctx context.Context, req *agentpb.CompleteUploadRequest, record store.UploadRecord) (bool, error) {
-						if err := authorizeAgentSession(req.Meta); err != nil {
-							return false, err
-						}
-						return d.CompleteLocalMockRecording(ctx, req, record)
-					},
-				})
-				if handlerErr != nil {
-					return handlerErr
-				}
-				eventHandler, eventErr := rpc.NewDispatcherEventServer(st, rpc.DispatcherEventServerOptions{
-					RequirePeer: true, PeerCertificateFingerprints: peerFingerprints,
-					AllowedAgentIDs: allowedAgentIDs, Now: time.Now,
-					LocalV3RecordingFailure: d.RecordLocalMockRecordingFailure,
-					LocalV3SessionCheck:     authorizeAgentSession,
-				})
-				if eventErr != nil {
-					return eventErr
-				}
-				dispatcherHandler := rpc.NewDispatcherServer(uploadHandler, eventHandler)
-				tlsConfig, tlsErr := rpc.LoadServerTLSConfig(cfg.MTLSCAFile, cfg.MTLSCertFile, cfg.MTLSKeyFile)
-				if tlsErr != nil {
-					return fmt.Errorf("configure Dispatcher gRPC mTLS: %w", tlsErr)
-				}
-				dispatcherListener, err = net.Listen("tcp", cfg.DispatcherGRPCListen)
-				if err != nil {
-					return fmt.Errorf("listen Dispatcher gRPC: %w", err)
-				}
-				dispatcherGRPC = grpc.NewServer(grpc.Creds(credentials.NewTLS(tlsConfig)), grpc.UnaryInterceptor(func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
-					if leaseCtx.Err() != nil {
-						return nil, status.Error(codes.Unavailable, "Dispatcher ownership is no longer active")
-					}
-					return handler(ctx, req)
-				}))
-				agentpb.RegisterAgentControlServiceServer(dispatcherGRPC, dispatcherHandler)
-				go func() {
-					if serveErr := dispatcherGRPC.Serve(dispatcherListener); serveErr != nil && !errors.Is(serveErr, grpc.ErrServerStopped) {
-						slog.Error("Dispatcher gRPC stopped", "error", serveErr)
-					}
-				}()
-				defer func() {
-					dispatcherGRPC.GracefulStop()
-					_ = dispatcherListener.Close()
-				}()
-			}
-			result := map[string]any{"role": "dispatcher", "mode": cfg.Mode, "db": cfg.DBPath}
-			if dispatcherGRPC != nil {
-				result["dispatcher_grpc"] = cfg.DispatcherGRPCListen
-			}
-			if connectedAgents != nil {
-				sessions := make([]map[string]any, 0, len(connectedAgents.sessions))
-				for _, session := range connectedAgents.sessions {
-					sessions = append(sessions, map[string]any{
-						"agent_id": session.AgentID, "cell_id": session.CellID,
-						"boot_id": session.BootID, "session_generation": session.SessionGeneration,
-					})
-				}
-				result["agent_sessions"] = sessions
-			}
-			if once {
-				if publisher == nil {
-					return errors.New("--once requires RABBITMQ_URL")
-				}
-				count, err := d.FlushOutbox(cmd.Context(), cfg.OutboxBatch)
-				if err != nil {
-					return err
-				}
-				result["published"] = count
-				return writeResult(result)
-			}
-			if broker == nil {
-				return errors.New("RABBITMQ_URL is required for the Dispatcher V3 runtime")
-			}
-			if connectedAgents == nil || len(connectedAgents.sessions) != 1 {
-				return errors.New("the Dispatcher V3 runtime requires exactly one configured Agent")
-			}
-			configClient, err := configread.NewClient(os.Getenv("DISPATCHER_CONFIG_READ_BASE_URL"), cfg.DispatcherID,
-				os.Getenv("DISPATCHER_SECRET_KEY"), &http.Client{Timeout: 15 * time.Second})
-			if err != nil {
-				return fmt.Errorf("configure Dispatcher config-read client: %w", err)
-			}
-			session := connectedAgents.sessions[0]
-			verifier := &dispatcher.AgentSIPConfigVerifier{
-				Probe: connectedAgents.coordinator, AgentID: session.AgentID, CellID: session.CellID,
-			}
-			runtime, err := dispatcher.NewLocalV01Runtime(d, broker, configClient, verifier, connectedAgents.coordinator)
-			if err != nil {
-				return err
-			}
-			if err := runtime.EnableMockAuthorizedOrigination(session.AgentID, connectedAgents.coordinator); err != nil {
-				return fmt.Errorf("enable isolated Mock Agent execution: %w", err)
-			}
-			result["queue_protocol"] = "v3"
-			if err := writeResult(result); err != nil {
-				return err
-			}
-			runtimeDone := make(chan error, 1)
-			go func() { runtimeDone <- runtime.Run(leaseCtx) }()
-			select {
-			case <-leaseCtx.Done():
-				if errors.Is(context.Cause(leaseCtx), errDispatcherIdentityLost) {
-					return context.Cause(leaseCtx)
-				}
-				return nil
-			case err := <-lease.Lost():
-				return fmt.Errorf("dispatcher lease lost: %w", err)
-			case err := <-runtimeDone:
-				if err != nil {
-					return err
-				}
-				if leaseCtx.Err() != nil {
-					return nil
-				}
-				return errors.New("Dispatcher V3 runtime stopped unexpectedly")
-			}
-		},
-	}
-	cmd.Flags().StringVar(&cfg.Mode, "mode", cfg.Mode, "mock, mixed, or real")
-	cmd.Flags().StringVar(&cfg.DBPath, "db", cfg.DBPath, "Dispatcher SQLite path")
-	cmd.Flags().StringVar(&configFile, "config", "", "required strict Dispatcher JSON configuration file")
-	cmd.Flags().StringVar(&cfg.RabbitURL, "rabbit-url", cfg.RabbitURL, "RabbitMQ URL")
-	cmd.Flags().IntVar(&cfg.OutboxBatch, "outbox-batch", cfg.OutboxBatch, "maximum outbox messages per run")
-	cmd.Flags().StringVar(&cfg.AgentEndpointsFile, "agent-endpoints-file", cfg.AgentEndpointsFile, "strict JSON file of Dispatcher-owned Agent endpoints")
-	cmd.Flags().BoolVar(&once, "once", false, "flush one outbox batch and exit")
-	return cmd
-}
-
-func parseCSVSet(raw string) map[string]struct{} {
-	result := make(map[string]struct{})
-	for _, item := range strings.Split(raw, ",") {
-		item = strings.TrimSpace(item)
-		if item != "" {
-			result[item] = struct{}{}
-		}
-	}
-	return result
-}
-
-func marshalResultForTest(value any) ([]byte, error) {
-	return json.Marshal(value)
-}
-
-func writeResult(value any) error {
-	data, err := marshalResultForTest(value)
-	if err != nil {
-		return err
-	}
-	_, err = os.Stdout.Write(append(data, '\n'))
-	return err
-}
-
-func signalContext() (context.Context, context.CancelFunc) {
-	return signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
-}

cmd/sip-go-agent/main_structure_test.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"testing"
+)
+
+func TestMainContainsOnlyProcessEntryPoints(t *testing.T) {
+	file, err := parser.ParseFile(token.NewFileSet(), "main.go", nil, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, declaration := range file.Decls {
+		function, ok := declaration.(*ast.FuncDecl)
+		if !ok {
+			continue
+		}
+		if function.Name.Name != "main" && function.Name.Name != "newRootCommand" {
+			t.Errorf("obsolete process entry or helper remains in main.go: %s", function.Name.Name)
+		}
+	}
+}

cmd/sip-go-agent/main_test.go

@@ -1,12 +1,8 @@
 package main
 
 import (
-	"bytes"
 	"strings"
 	"testing"
-
-	"git.ipao.vip/rogee/go-sip/contracts"
-	"git.ipao.vip/rogee/go-sip/internal/contract"
 )
 
 func TestRootHasExplicitRoles(t *testing.T) {
@@ -34,57 +30,3 @@ func TestRootDispatcherRejectsOldConfigFlag(t *testing.T) {
 		t.Fatalf("obsolete Dispatcher configuration entry was admitted: %v", err)
 	}
 }
-
-func TestWriteResultIsJSON(t *testing.T) {
-	var b bytes.Buffer
-	data, err := marshalResultForTest(map[string]string{"status": "ok"})
-	if err != nil || len(data) == 0 {
-		t.Fatalf("data=%q err=%v", data, err)
-	}
-	b.Write(data)
-	if b.Len() == 0 {
-		t.Fatal("empty result")
-	}
-}
-
-func TestValidateCallAISnapshotAllowsASROnly(t *testing.T) {
-	raw, err := contracts.Read("examples/agent-version-asr-only.json")
-	if err != nil {
-		t.Fatal(err)
-	}
-	snapshot, err := validateCallAISnapshot(raw)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if snapshot.Mode != "asr_only" {
-		t.Fatalf("snapshot mode=%q, want asr_only", snapshot.Mode)
-	}
-}
-
-func TestBuildCallEndpointUsesArtifactRoute(t *testing.T) {
-	raw, err := contracts.Files.ReadFile("local/v0.3/examples/static-cell-artifact-real-v2.json")
-	if err != nil {
-		t.Fatal(err)
-	}
-	artifact, err := contract.ValidateStaticArtifact(raw, contract.StaticArtifactExpectation{CellID: "cell-single", Mode: "real"})
-	if err != nil {
-		t.Fatal(err)
-	}
-	endpoint, err := buildCallEndpoint(artifact, "provider-third", "15830461047")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if endpoint != "PJSIP/mka75515830461047@provider-third" {
-		t.Fatalf("endpoint=%q", endpoint)
-	}
-	media, err := mediaForCall(artifact, "provider-third")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if media.Format != "alaw" || media.SampleRateHz != 8000 || media.PayloadType != 8 {
-		t.Fatalf("media=%+v", media)
-	}
-	if _, err := buildCallEndpoint(artifact, "provider-third", "10000000000"); err == nil {
-		t.Fatal("disallowed target was accepted")
-	}
-}

cmd/sip-go-agent/upload_retry_command.go

@@ -1,53 +0,0 @@
-package main
-
-import (
-	"errors"
-	"time"
-
-	"git.ipao.vip/rogee/go-sip/internal/agent"
-	"git.ipao.vip/rogee/go-sip/internal/config"
-	"git.ipao.vip/rogee/go-sip/internal/rpc"
-	"github.com/spf13/cobra"
-)
-
-func newUploadRetryCommand() *cobra.Command {
-	cfg, configErr := config.FromEnv()
-	var uploadID, requestID, path string
-	cmd := &cobra.Command{
-		Use: "upload-retry", Short: "explicitly request one new grant for an unsuccessful upload; never dial",
-		Args: cobra.NoArgs,
-		RunE: func(cmd *cobra.Command, _ []string) (returnErr error) {
-			if configErr != nil {
-				return configErr
-			}
-			if uploadID == "" || requestID == "" || path == "" {
-				return errors.New("--upload-id, --request-id and --file are required")
-			}
-			if err := cfg.Validate("agent"); err != nil {
-				return err
-			}
-			if cfg.DispatcherGRPCEndpoint == "" || cfg.DispatcherGRPCServerName == "" {
-				return errors.New("Dispatcher gRPC endpoint and server name are required")
-			}
-			spool, err := agent.NewSpool(cfg.SpoolRoot, time.Now)
-			if err != nil {
-				return err
-			}
-			client, err := rpc.DialFromFiles(cfg.DispatcherGRPCEndpoint, cfg.MTLSCAFile, cfg.MTLSCertFile, cfg.MTLSKeyFile, cfg.DispatcherGRPCServerName)
-			if err != nil {
-				return err
-			}
-			defer func() { returnErr = errors.Join(returnErr, client.Close()) }()
-			record, err := retryUploadRecording(cmd.Context(), cfg, client, agent.UploadClient{Now: time.Now}, spool, uploadID, requestID, path)
-			if err != nil {
-				return err
-			}
-			return writeResult(map[string]any{"upload_id": record.UploadID, "notification_state": record.State, "bytes": record.Result.SizeBytes, "sha256": record.Result.SHA256})
-		},
-	}
-	cmd.Flags().StringVar(&cfg.SpoolRoot, "spool", cfg.SpoolRoot, "existing Agent spool root")
-	cmd.Flags().StringVar(&uploadID, "upload-id", "", "existing failed or uncertain upload identity")
-	cmd.Flags().StringVar(&requestID, "request-id", "", "explicit new canonical UUID v4; never reuse a consumed request")
-	cmd.Flags().StringVar(&path, "file", "", "retained original recording; content must match its persisted identity")
-	return cmd
-}

docs/evidence/saas-dispatcher-implementation.md

@@ -97,6 +97,7 @@
 ## P07 单一入口清理（执行中）
 
 - 数据归属根因：现行 SQLite 的 `CloseAdmission` 原本只按传入的 Dispatcher ID 插入或更新状态；把已有未投递 outbox 的 D1 数据库交给 D2 启动时，会错误建立第二个归属状态。新测试先证实该问题，再要求同一事务内检查既有归属，冲突时返回身份错误且保留原归属和未确认 outbox；本轮不迁移、不删除任何既有数据库。旧入口及名称残留仍须分批清理，不能因该测试通过便宣称 P07 完成。
+- CLI 去旧入口：`main.go` 先由 AST 回归测试证实还有 14 个与进程入口无关的函数，随后收敛为 `main` 与唯一 `newRootCommand`；未注册的旧 Agent/Dispatcher 入口及旧上传重试命令直接移除，根命令仍仅提供当前 Agent/Dispatcher 两个显式角色。根角色/废弃参数、非 Mock 预检已有现行命令测试覆盖；原 SQLite 身份/未投递回执保护已由现行 Store 回归接管，线路前缀与 ASR-only 的行为由现行策略及 AI 测试覆盖。不迁移或删除任何实际 SQLite、spool 或 outbox。其余旧模块、契约、引用及代次命名尚待清查，不能将本批当作 P07 整体完成。
 
 ## 验收台账