diff --git a/cmd/sip-go-agent/dispatcher_command_test.go b/cmd/sip-go-agent/dispatcher_command_test.go deleted file mode 100644 index bf60f57..0000000 --- a/cmd/sip-go-agent/dispatcher_command_test.go +++ /dev/null @@ -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) - } - }) - } -} diff --git a/cmd/sip-go-agent/dispatcher_config_test.go b/cmd/sip-go-agent/dispatcher_config_test.go index e6a014b..c5c6ccd 100644 --- a/cmd/sip-go-agent/dispatcher_config_test.go +++ b/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) - } -} diff --git a/cmd/sip-go-agent/main.go b/cmd/sip-go-agent/main.go index 530b938..bb6b791 100644 --- a/cmd/sip-go-agent/main.go +++ b/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) -} diff --git a/cmd/sip-go-agent/main_structure_test.go b/cmd/sip-go-agent/main_structure_test.go new file mode 100644 index 0000000..ed27477 --- /dev/null +++ b/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) + } + } +} diff --git a/cmd/sip-go-agent/main_test.go b/cmd/sip-go-agent/main_test.go index d5c2c0a..54cb67f 100644 --- a/cmd/sip-go-agent/main_test.go +++ b/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") - } -} diff --git a/cmd/sip-go-agent/upload_retry_command.go b/cmd/sip-go-agent/upload_retry_command.go deleted file mode 100644 index 0f72aea..0000000 --- a/cmd/sip-go-agent/upload_retry_command.go +++ /dev/null @@ -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 -} diff --git a/docs/evidence/saas-dispatcher-implementation.md b/docs/evidence/saas-dispatcher-implementation.md index 5ef80d9..c0ca4d4 100644 --- a/docs/evidence/saas-dispatcher-implementation.md +++ b/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 整体完成。 ## 验收台账