package rpc import ( "context" "encoding/hex" "log/slog" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) const authorizedOriginationVersion = "agent-authorized-origination.v0.1" // ExecuteAuthorized only runs an explicit mock adapter. D has already made and // durably claimed the final dial decision. A enforces the D-issued exclusive // deadline and session identity, but never recalculates business policy. func (s *Server) ExecuteAuthorized(ctx context.Context, req *agentpb.ExecuteAuthorizedRequest) (*agentpb.ExecuteAuthorizedResponse, error) { if req == nil || req.Meta == nil || req.Binding == nil { return nil, status.Error(codes.InvalidArgument, "request metadata and execution binding are required") } if s.mode != "mock" || s.mockAuthorizedOriginate == nil { return nil, status.Error(codes.FailedPrecondition, "authorized origination requires an explicit mock adapter") } if err := s.authorize(ctx, req.Meta); err != nil { return nil, err } if err := requireIdempotency(req.Meta); err != nil { return nil, err } binding := req.Binding if req.SchemaVersion != authorizedOriginationVersion || binding.ExecutionId == "" || binding.TenantId == "" || binding.TenantKey == "" || binding.TaskId == "" || binding.TaskItemId == "" || req.SelectedTrunkId == "" || req.CallerId == "" || req.Callee == "" || req.RingTimeoutMs <= 0 || req.MaxCallDurationMs <= 0 || req.DialBeforeUnixMs <= 0 || !validLowerSHA256(req.BoundSnapshotSha256) { return nil, status.Error(codes.InvalidArgument, "invalid authorized origination version, binding or dial decision") } digest := messageDigest(req) key := s.operationKey(req.Meta) s.mu.Lock() if s.executionErr != nil { s.mu.Unlock() return nil, status.Errorf(codes.Internal, "execution journal unavailable: %v", s.executionErr) } if previous, exists := s.operations[key]; exists { if previous.digest != digest { s.mu.Unlock() return &agentpb.ExecuteAuthorizedResponse{Receipt: s.receipt(req.Meta, agentpb.ResultCode_RESULT_CODE_CONFLICT, agentpb.FailureCode_FAILURE_CODE_ABORTED, "idempotency key content conflict", false)}, nil } entry := s.executions[binding.ExecutionId] if entry == nil { s.mu.Unlock() return nil, status.Error(codes.Internal, "persisted operation lacks execution state") } response := &agentpb.ExecuteAuthorizedResponse{Receipt: proto.Clone(previous.receipt).(*agentpb.OperationReceipt), State: entry.state} s.mu.Unlock() return response, nil } if _, exists := s.executions[binding.ExecutionId]; exists { s.mu.Unlock() return &agentpb.ExecuteAuthorizedResponse{Receipt: s.receipt(req.Meta, agentpb.ResultCode_RESULT_CODE_CONFLICT, agentpb.FailureCode_FAILURE_CODE_ABORTED, "execution already exists", false)}, nil } if s.now().UnixMilli() >= req.DialBeforeUnixMs { s.mu.Unlock() return nil, status.Error(codes.FailedPrecondition, "Dispatcher-issued dial deadline expired") } // Persist an unknown one-shot attempt BEFORE contacting the mock adapter. // After a crash or lost response, replay/query cannot cause a second dial. s.executions[binding.ExecutionId] = &executionRecord{ executeDigest: digest, binding: proto.Clone(binding).(*agentpb.ExecutionBinding), state: agentpb.ExecutionState_EXECUTION_STATE_UNKNOWN, taskRevision: binding.TaskRevision, unknown: true, callState: "mock_originating_unknown", } pending := s.receipt(req.Meta, agentpb.ResultCode_RESULT_CODE_UNKNOWN, agentpb.FailureCode_FAILURE_CODE_UNAVAILABLE, "mock originate pending; do not retry", false) s.operations[key] = operationRecord{digest: digest, receipt: pending} if err := s.persistExecutionJournalLocked(); err != nil { s.mu.Unlock() return nil, err } s.mu.Unlock() // A newly closed deadline between durable claim and adapter invocation must // not dial. This is a technical check of D's instruction, not a new policy. expired := s.now().UnixMilli() >= req.DialBeforeUnixMs var dialErr error if !expired { dialErr = s.mockAuthorizedOriginate(ctx, proto.Clone(req).(*agentpb.ExecuteAuthorizedRequest)) } s.mu.Lock() defer s.mu.Unlock() record := s.executions[binding.ExecutionId] var result agentpb.ResultCode var failure agentpb.FailureCode var detail string switch { case expired: record.state, record.unknown, record.callState = agentpb.ExecutionState_EXECUTION_STATE_TERMINAL, false, "mock_deadline_closed_without_dial" record.terminalObservedAtUnixMs = s.now().UnixMilli() result, failure, detail = agentpb.ResultCode_RESULT_CODE_REJECTED, agentpb.FailureCode_FAILURE_CODE_FAILED_PRECONDITION, "dial deadline expired before mock adapter" case dialErr != nil: slog.Error("mock originate failed; execution remains unknown and cannot be retried", "execution_id", binding.ExecutionId, "error", dialErr) record.state, record.unknown, record.callState = agentpb.ExecutionState_EXECUTION_STATE_UNKNOWN, true, "mock_originating_unknown" result, failure, detail = agentpb.ResultCode_RESULT_CODE_UNKNOWN, agentpb.FailureCode_FAILURE_CODE_UNAVAILABLE, "mock originate outcome unknown; reconcile instead of retry" default: record.state, record.unknown, record.callState = agentpb.ExecutionState_EXECUTION_STATE_TERMINAL, false, "mock_no_answer" record.terminalObservedAtUnixMs = s.now().UnixMilli() result, failure, detail = agentpb.ResultCode_RESULT_CODE_APPLIED, agentpb.FailureCode_FAILURE_CODE_UNSPECIFIED, "" } receipt := s.receipt(req.Meta, result, failure, detail, false) s.operations[key] = operationRecord{digest: digest, receipt: receipt} if err := s.persistExecutionJournalLocked(); err != nil { return nil, err } return &agentpb.ExecuteAuthorizedResponse{Receipt: receipt, State: record.state}, nil } func validLowerSHA256(value string) bool { if len(value) != 64 { return false } decoded, err := hex.DecodeString(value) return err == nil && hex.EncodeToString(decoded) == value }